feat: index historical votes

.env-example

@@ -3,3 +3,4 @@ REGEN_RPC=
 REGEN_API=
 # START_BLOCK_OVERRIDE=
 # ONLY_INDEX_SPECIFIC_BLOCKS=
+# REGEN_IS_ARCHIVE_NODE=false

index_class_issuers.py

@@ -2,15 +2,19 @@
 import os
 import textwrap
 import requests
-from utils import PollingProcess, events_to_process
+from utils import is_archive_node, PollingProcess, events_to_process
 
 logger = logging.getLogger(__name__)
 
 
 def fetch_class_issuers(height, class_id):
+    if is_archive_node():
+        headers = {"x-cosmos-block-height": str(height)}
+    else:
+        headers = None
     resp = requests.get(
         f"{os.environ['REGEN_API']}/regen/ecocredit/v1/classes/{class_id}/issuers",
-        headers={"x-cosmos-block-height": str(height)},
+        headers=headers,
     )
     resp.raise_for_status()
     return resp.json()["issuers"]

index_votes.py

@@ -0,0 +1,129 @@
+import logging
+import os
+import textwrap
+from psycopg2.errors import ForeignKeyViolation
+import requests
+from utils import is_archive_node, PollingProcess, events_to_process
+
+logger = logging.getLogger(__name__)
+
+
+def fetch_votes_by_proposal(height, proposal_id):
+    if is_archive_node():
+        headers = {"x-cosmos-block-height": str(height)}
+    else:
+        headers = None
+    # Currently EventVote only contains proposal id
+    # Eventually EventVote may contain proposal id and voter address
+    # At which point we could get the vote with this endpoint:
+    # /cosmos/group/v1/vote_by_proposal_voter/{proposal_id}/{voter}
+    # Ref: https://github.com/regen-network/indexer/pull/38#discussion_r1310958235
+    resp = requests.get(
+        f"{os.environ['REGEN_API']}/cosmos/group/v1/votes_by_proposal/{proposal_id}",
+        headers=headers,
+    )
+    resp.raise_for_status()
+    return resp.json()["votes"]
+
+
+def _index_votes(pg_conn, _client, _chain_num):
+    with pg_conn.cursor() as cur:
+        for event in events_to_process(
+            cur,
+            "votes",
+        ):
+            (
+                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
+            logger.debug(normalize)
+            votes = fetch_votes_by_proposal(
+                normalize["block_height"], normalize["proposal_id"]
+            )
+            logger.debug(votes)
+            insert_text = textwrap.dedent(
+                """
+            INSERT INTO votes (
+                type,
+                block_height,
+                tx_idx,
+                msg_idx,
+                chain_num,
+                timestamp,
+                tx_hash,
+                proposal_id,
+                voter,
+                option,
+                metadata,
+                submit_time
+            ) VALUES (
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s,
+                %s
+            );"""
+            ).strip("\n")
+            with pg_conn.cursor() as _cur:
+                for vote in votes:
+                    try:
+                        row = (
+                            normalize["type"],
+                            normalize["block_height"],
+                            normalize["tx_idx"],
+                            normalize["msg_idx"],
+                            normalize["chain_num"],
+                            normalize["timestamp"],
+                            normalize["tx_hash"],
+                            normalize["proposal_id"],
+                            vote["voter"],
+                            vote["option"],
+                            vote["metadata"],
+                            vote["submit_time"],
+                        )
+                        _cur.execute(
+                            insert_text,
+                            row,
+                        )
+                        logger.debug(_cur.statusmessage)
+                        pg_conn.commit()
+                        logger.info("vote inserted..")
+                    except ForeignKeyViolation as exc:
+                        logger.debug(exc)
+                        pg_conn.rollback()
+                        # since we know all votes for this proposal will fail we exit the loop
+                        break
+
+
+def index_votes():
+    p = PollingProcess(
+        target=_index_votes,
+        sleep_secs=1,
+    )
+    p.start()

main.py

@@ -7,6 +7,7 @@
 from index_retires import index_retires
 from index_proposals import index_proposals
 from index_class_issuers import index_class_issuers
+from index_votes import index_votes
 
 load_dotenv()
 
@@ -36,3 +37,4 @@
     index_retires()
     index_proposals()
     index_class_issuers()
+    index_votes()

sql/V1_12__votes.sql

@@ -0,0 +1,39 @@
+ALTER TABLE IF EXISTS proposals
+DROP CONSTRAINT IF EXISTS proposals_proposal_id_ux;
+
+ALTER TABLE IF EXISTS proposals
+ADD CONSTRAINT proposals_proposal_id_ux UNIQUE (chain_num, proposal_id);
+
+CREATE TABLE IF NOT EXISTS
+  votes (
+    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,
+    voter TEXT NOT NULL,
+    OPTION TEXT NOT NULL,
+    metadata TEXT NOT NULL,
+    submit_time timestamptz 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,
+    FOREIGN KEY (chain_num, proposal_id) REFERENCES proposals (chain_num, proposal_id)
+  );
+
+DROP INDEX IF EXISTS votes_proposal_id_chain_num_idx;
+
+CREATE INDEX IF NOT EXISTS votes_proposal_id_chain_num_idx ON votes (proposal_id, chain_num);

sql/run_all_migrations.sh

@@ -15,3 +15,4 @@ psql -c "\i V1_8__index_proposal_id.sql" $DATABASE_URL
 psql -c "\i V1_9__all_ecocredit_txes.sql" $DATABASE_URL
 psql -c "\i V1_10__class_issuers.sql" $DATABASE_URL
 psql -c "\i V1_11__class_issuers_indexes.sql" $DATABASE_URL
+psql -c "\i V1_12__votes.sql" $DATABASE_URL

utils.py

@@ -108,6 +108,7 @@ def run(self):
         "regen.ecocredit.v1.EventCreateClass",
         "regen.ecocredit.v1.EventUpdateClassIssuers",
     ],
+    "votes": ["cosmos.group.v1.EventVote"],
 }
 
 
@@ -145,3 +146,12 @@ def events_to_process(cur, index_table_name):
     # this is how key and value are put into their own column
     for _, g in groupby(cur, lambda x: f"{x[1]}-{x[2]}-{x[3]}"):
         yield list(g)
+
+
+def is_archive_node():
+    # since the indexer is intended to run against archive nodes,
+    # assume that by default the node is an archive node.
+    value = os.environ.get("REGEN_IS_ARCHIVE_NODE", "true").lower()
+    if value not in ["true", "false"]:
+        raise ValueError("REGEN_IS_ARCHIVE_NODE must be true or false")
+    return value == "true"