feat: index proposals

index_proposals.py

@@ -0,0 +1,120 @@
+import logging
+import os
+import textwrap
+from psycopg2.extras import Json
+import requests
+from utils import PollingProcess, events_to_process
+
+logger = logging.getLogger(__name__)
+
+
+def fetch_proposal(height, proposal_id):
+    resp = requests.get(
+        f"{os.environ['REGEN_API']}/cosmos/group/v1/proposal/{proposal_id}",
+        headers={"x-cosmos-block-height": str(height)},
+    )
+    resp.raise_for_status()
+    return resp.json()["proposal"]
+
+
+def _index_proposals(pg_conn, _client, _chain_num):
+    with pg_conn.cursor() as cur:
+        for event in events_to_process(
+            cur,
+            "proposals",
+        ):
+            (type, block_height, tx_idx, msg_idx, _, _, chain_num, timestamp, tx_hash) = event[0]
+            normalize = {}
+            normalize["type"] = type
+            normalize["block_height"] = block_height
+            normalize["tx_idx"] = tx_idx
+            normalize["msg_idx"] = msg_idx
+            normalize["chain_num"] = chain_num
+            normalize["timestamp"] = timestamp
+            normalize["tx_hash"] = tx_hash
+            for entry in event:
+                (_, _, _, _, key, value, _, _, _) = entry
+                value = value.strip('"')
+                normalize[key] = value
+            proposal = fetch_proposal(
+                normalize["block_height"] - 1, normalize["proposal_id"]
+            )
+            row = (
+                normalize["type"],
+                normalize["block_height"],
+                normalize["tx_idx"],
+                normalize["msg_idx"],
+                normalize["chain_num"],
+                normalize["timestamp"],
+                normalize["tx_hash"],
+                proposal["id"],
+                proposal["status"],
+                proposal["group_policy_address"],
+                proposal["metadata"],
+                proposal["proposers"],
+                proposal["submit_time"],
+                proposal["group_version"],
+                proposal["group_policy_version"],
+                Json(proposal["final_tally_result"]),
+                proposal["voting_period_end"],
+                proposal["executor_result"],
+                Json(proposal["messages"]),
+            )
+            insert_text = textwrap.dedent("""
+            INSERT INTO proposals (
+                type,
+                block_height,
+                tx_idx,
+                msg_idx,
+                chain_num,
+                timestamp,
+                tx_hash,
+                proposal_id,
+                status,
+                group_policy_address,
+                metadata,
+                proposers,
+                submit_time,
+                group_version,
+                group_policy_version,
+                final_tally_result,
+                voting_period_end,
+                executor_result,
+                messages
+            ) VALUES (
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s
+            );""").strip("\n")
+            with pg_conn.cursor() as _cur:
+                _cur.execute(
+                    insert_text,
+                    row,
+                )
+                logger.debug(_cur.statusmessage)
+                pg_conn.commit()
+                logger.info("proposal inserted...")
+
+
+def index_proposals():
+    p = PollingProcess(
+        target=_index_proposals,
+        sleep_secs=1,
+    )
+    p.start()

main.py

@@ -5,6 +5,7 @@
 import sentry_sdk
 from index_blocks import index_blocks
 from index_retires import index_retires
+from index_proposals import index_proposals
 
 load_dotenv()
 
@@ -32,3 +33,4 @@
 if __name__ == "__main__":
     index_blocks()
     index_retires()
+    index_proposals()

sql/V1_5__add_proposals_table.sql

@@ -0,0 +1,37 @@
+CREATE TABLE IF NOT EXISTS
+  proposals (
+    TYPE TEXT NOT NULL,
+    block_height BIGINT NOT NULL,
+    tx_idx SMALLINT NOT NULL,
+    msg_idx SMALLINT NOT NULL,
+    chain_num SMALLINT NOT NULL,
+    TIMESTAMP timestamptz,
+    tx_hash TEXT NOT NULL,
+
+    proposal_id BIGINT NOT NULL,
+    status TEXT NOT NULL,
+    group_policy_address TEXT NOT NULL,
+    metadata TEXT NOT NULL,
+    proposers TEXT[] NOT NULL,
+    submit_time timestamptz,
+    group_version BIGINT NOT NULL,
+    group_policy_version BIGINT NOT NULL,
+    final_tally_result JSONB NOT NULL,
+    voting_period_end timestamptz NOT NULL,
+    executor_result TEXT NOT NULL,
+    messages JSONB NOT NULL,
+
+    PRIMARY KEY (
+      chain_num,
+      block_height,
+      tx_idx,
+      msg_idx
+    ),
+    FOREIGN KEY (
+      chain_num,
+      block_height,
+      tx_idx,
+      msg_idx,
+      TYPE
+    ) REFERENCES msg_event
+  );

sql/run_all_migrations.sh

@@ -8,4 +8,5 @@ psql -c "\i V1_1__fix_msg_attr.sql" $DATABASE_URL
 psql -c "\i V1_2__add_retirements_table.sql" $DATABASE_URL
 psql -c "\i V1_3__add_msg_event_attr_type_idx.sql" $DATABASE_URL
 psql -c "\i V1_4__retirements_owner_idx.sql" $DATABASE_URL
+psql -c "\i V1_5__add_proposals_table.sql" $DATABASE_URL
 psql -c "\i V1_6__add_tx_hash.sql" $DATABASE_URL

utils.py

@@ -103,6 +103,7 @@ def run(self):
         "regen.ecocredit.v1.EventRetire",
         "regen.ecocredit.v1alpha1.EventRetire",
     ],
+    "proposals": ["cosmos.group.v1.EventProposalPruned"],
 }