Main logic for saving data & events to storage

Author: Reecepbcups

.gitignore

@@ -0,0 +1,3 @@
+blocks/*.json
+latest_height.txt
+__pycache__/

main.py

@@ -0,0 +1,104 @@
+import json
+import os
+
+import rel
+import websocket
+
+from util import (
+    decode_txs,
+    get_block_events,
+    get_block_height,
+    get_block_txs,
+    get_unique_event_addresses,
+    remove_useless_data,
+    update_latest_height,
+)
+
+# TODO: Save Txs & events to postgres?
+# maybe a redis cache as well so other people can subscribe to redis for events?
+
+# https://docs.tendermint.com/v0.34/rpc/
+RPC_IP = "15.204.143.232:26657"
+RPC_URL = f"ws://{RPC_IP}/websocket"
+WALLET_PREFIX = "juno1"
+COSMOS_BINARY_FILE = "junod"
+
+current_dir = os.path.dirname(os.path.realpath(__file__))
+blocks = os.path.join(current_dir, "blocks")
+os.makedirs(blocks, exist_ok=True)
+
+
+# TODO: get current height. If there is a difference, then we need to query the RPC for the missing blocks.
+latest_height_file = os.path.join(current_dir, "latest_height.txt")
+latest_height = -1
+if os.path.exists(latest_height_file):
+    with open(latest_height_file, "r") as f:
+        latest_height = int(f.read())
+print(f"Latest Height: {latest_height}")
+
+
+def on_message(ws, message):
+    msg = dict(json.loads(message))
+
+    if msg.get("result") == {}:
+        print("Subscribed to New Block...")
+        return
+
+    block_data = msg.get("result", {})
+
+    # Get height & save latest to file
+    block_height = get_block_height(block_data)
+    update_latest_height(latest_height_file, block_height)
+
+    # Gets block transactions, decodes them to JSON, and saves them to the block_data
+    block_txs = get_block_txs(block_data)
+    decoded_txs = decode_txs(COSMOS_BINARY_FILE, block_txs)
+    block_data["data"]["value"]["block"]["data"]["txs"] = decoded_txs
+
+    # Gets unique addresses from events (users/contracts interacted with during this time frame)
+    # Useful for cache solutions. So if a user does not have any changes here, then we can keep them cached longer
+    block_events = get_block_events(block_data)
+    unique_event_addresses = get_unique_event_addresses(WALLET_PREFIX, block_events)
+    block_data["events"]["all_unique_event_addresses"] = list(unique_event_addresses)
+
+    # Removes useless events we do not need to cache which take up lots of space
+    updated_data = remove_useless_data(block_data)
+
+    # Saves data to a file
+    # TODO: learn postgres and save with relations
+    with open(os.path.join(blocks, f"{block_height}.json"), "w") as f:
+        f.write(json.dumps(updated_data))
+
+
+def on_error(ws, error):
+    print("error", error)
+
+
+def on_close(ws, close_status_code, close_msg):
+    print("### closed ###")
+
+
+def on_open(ws):
+    print("Opened connection")
+    ws.send(
+        '{"jsonrpc": "2.0", "method": "subscribe", "params": ["tm.event=\'NewBlock\'"], "id": 1}'
+    )
+    print("Sent subscribe request")
+
+
+# from websocket import create_connection
+if __name__ == "__main__":
+    websocket.enableTrace(False)  # toggle to show or hide output
+    ws = websocket.WebSocketApp(
+        f"{RPC_URL}",
+        on_open=on_open,
+        on_message=on_message,
+        on_error=on_error,
+        on_close=on_close,
+    )
+
+    ws.run_forever(
+        dispatcher=rel, reconnect=5
+    )  # Set dispatcher to automatic reconnection, 5 second reconnect delay if connection closed unexpectedly
+    rel.signal(2, rel.abort)  # Keyboard Interrupt
+    rel.dispatch()

util.py

@@ -0,0 +1,87 @@
+import json
+import os
+
+current_dir = os.path.dirname(os.path.realpath(__file__))
+
+
+def __run_decode(cosmos_binary: str, tx: str) -> dict:
+    # check for max len tx (store code breaks this for CLI usage on linux)
+    if len(tx) > 32766:
+        print("TX too long. Skipping...")
+        return {}
+
+    res = os.popen(f"{cosmos_binary} tx decode {tx} --output json").read()
+    return json.loads(res)
+
+
+def decode_txs(COSMOS_BINARY_FILE: str, block_txs: dict) -> list:
+    decoded_txs = []
+
+    # iterate through each and convert to json with junod
+    for tx in block_txs:
+        tx = __run_decode(COSMOS_BINARY_FILE, tx)
+        if isinstance(tx, dict) and len(tx) > 0:
+            decoded_txs.append(tx)
+
+    return decoded_txs
+
+def get_block_txs(block_data: dict) -> list:
+    return (
+        block_data.get("data", {})
+        .get("value", {})
+        .get("block", {})
+        .get("data", {})
+        .get("txs", [])
+    )
+
+
+def get_block_events(block_data: dict) -> dict:
+    return block_data.get("events", {})
+
+
+def get_block_height(block_data: dict) -> int:
+    height = block_data["data"]["value"]["block"]["header"]["height"]
+    print(f"Block Height: {height}")
+
+    return height
+
+
+def update_latest_height(latest_height_file: str, height: int) -> None:
+    with open(latest_height_file, "w") as f:
+        f.write(str(height))
+
+
+def get_unique_event_addresses(wallet_prefix: str, block_events: dict) -> list[str]:
+    # any address which had some action in the block
+    event_addresses: list[str] = []
+
+    for event_key, value in block_events.items():
+        if not isinstance(value, list):
+            continue
+
+        for v in value:
+            if isinstance(v, str) and v.startswith(wallet_prefix):
+                if v not in event_addresses:
+                    event_addresses.append(v)
+
+    return event_addresses
+
+
+def remove_useless_data(block_data: dict) -> dict:
+    # remove result_begin_block in the value section
+    del block_data["data"]["value"]["result_begin_block"]
+    del block_data["data"]["value"]["result_end_block"]
+    del block_data["data"]["value"]["block"]["last_commit"]["signatures"]
+
+    # remove usless events for us
+    del block_data["events"]["commission.amount"]
+    del block_data["events"]["commission.validator"]
+    del block_data["events"]["rewards.amount"]
+    del block_data["events"]["rewards.validator"]
+    del block_data["events"]["coin_spent.amount"]
+    del block_data["events"]["coin_received.receiver"]
+    del block_data["events"]["proposer_reward.amount"]
+    del block_data["events"]["mint.amount"]
+    del block_data["events"]["coinbase.amount"]
+
+    return block_data