add veAllocate (#910)

* add veAllocate

Author: alexcos20

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.4.2
+current_version = 4.5.1
 commit = True
 tag = True
 

README.md

@@ -63,6 +63,7 @@ The events monitor runs continuously to retrieve and index the chain Metadata. I
 - a Decryptor class that handles decompression and decryption on the chain data, through communication with Provider
 - a set of `ALLOWED_PUBLISHERS`, if such a restriction exists. You can set a limited number of allowed publisher addresses using this env variable.
 - a Purgatory, based on the `ASSET_PURGATORY_URL` and `ACCOUNT_PURGATORY_URL` env variables. These mark some assets as being in purgatory (`"isInPurgatory": True`), enabling restrictions for some assets or accounts.
+- a VeAllocate, based on the `VEALLOCATE_URL` and `VEALLOCATE_UPDATE_INTERVAL` env variables. This updates the veAllocation for datasets.
 - start blocks, if such defined using `BFACTORY_BLOCK` and `METADATA_CONTRACT_BLOCK`. These start blocks are coroborated with the last stored blocks per Elasticsearch, to avoid indexing multiple times
 
 The EventsMonitor processes block chunks as defined using `BLOCKS_CHUNK_SIZE`. For each block, it retrieves all `MetadataCreated` and `MetadataUpdated` events, and these events are processed inside the `MetadataCreatedProcessor` and `MetadataUpdatedProcessor` classes. These processors run the following flow:
@@ -142,6 +143,12 @@ ACCOUNT_PURGATORY_URL
 # Customise purgatory update (refresh) time (in number of minutes)
 PURGATORY_UPDATE_INTERVAL
 
+# URL for getting the veAllocation list. If not exists, the veAllocate will not be processed. Possible values are: https://df-sql.oceandao.org/nftinfo for mainnet and https://test-df-sql.oceandao.org/nftinfo for goerli, because veOCEAN is deployed only on this networks. All other networks SHOULD NOT HAVE this defined.  The list should be formatted as a list of dictionaries containing chainID,nft_addr and ve_allocated
+VEALLOCATE_URL
+
+# Customise veAllocate update (refresh) time (in number of minutes)
+VEALLOCATE_UPDATE_INTERVAL
+
 # The URL of the RBAC Permissions Server. If set, Aquarius will check permissions with RBAC. Leave empty/unset to skip RBAC permission checks.
 RBAC_SERVER_URL
 

aquarius/__init__.py

@@ -9,5 +9,5 @@
 __author__ = """OceanProtocol"""
 # fmt: off
 # bumpversion needs single quotes
-__version__ = '4.4.2'
+__version__ = '4.5.1'
 # fmt: on

aquarius/events/events_monitor.py

@@ -27,6 +27,7 @@
     TokenURIUpdatedProcessor,
 )
 from aquarius.events.purgatory import Purgatory
+from aquarius.events.ve_allocate import VeAllocate
 from aquarius.events.util import (
     get_metadata_start_block,
     get_defined_block,
@@ -101,6 +102,10 @@ def __init__(self, web3, config_file):
             else None
         )
 
+        self.ve_allocate = (
+            VeAllocate(self._es_instance) if (os.getenv("VEALLOCATE_URL")) else None
+        )
+
         self.retry_mechanism = RetryMechanism(
             config_file, self._es_instance, self._retries_db_index, self.purgatory
         )
@@ -146,6 +151,12 @@ def do_run_monitor(self):
             except (KeyError, Exception) as e:
                 logger.error(f"Error updating purgatory list: {str(e)}.")
 
+        if self.ve_allocate:
+            try:
+                self.ve_allocate.update_lists()
+            except (KeyError, Exception) as e:
+                logger.error(f"Error updating ve_allocate list: {str(e)}.")
+
     def process_current_blocks(self):
         """Process all blocks from the last processed block to the current block."""
         process_queue = strtobool(os.getenv("PROCESS_RETRY_QUEUE", "0"))

aquarius/events/processors.py

@@ -94,7 +94,11 @@ def add_aqua_data(self, record):
         order_count, price = get_number_orders_price(
             self.dt_contract.address, self.block, self._chain_id
         )
-        record[AquariusCustomDDOFields.STATS] = {"orders": order_count, "price": price}
+        record[AquariusCustomDDOFields.STATS] = {
+            "allocated": 0,
+            "orders": order_count,
+            "price": price,
+        }
 
         return record, block_time
 

aquarius/events/ve_allocate.py

@@ -0,0 +1,82 @@
+#
+# Copyright 2021 Ocean Protocol Foundation
+# SPDX-License-Identifier: Apache-2.0
+#
+import json
+import logging
+import os
+from datetime import datetime
+
+import elasticsearch
+import requests
+
+from aquarius.events.util import make_did
+
+logger = logging.getLogger(__name__)
+
+
+class VeAllocate:
+    def __init__(self, es_instance):
+        self.update_time = None
+        self._es_instance = es_instance
+
+    def retrieve_new_list(self, env_var):
+        """
+        :param env_var: Url of the file containing purgatory list.
+        :return: Object as follows: {...('<did>', '<reason>'),...}
+        """
+        response = requests.post(os.getenv(env_var))
+
+        if response.status_code == requests.codes.ok:
+            logger.info(
+                f"veAllocate: Successfully retrieved list from {env_var} env var."
+            )
+            return {
+                (a["nft_addr"], a["ve_allocated"], a["chainID"])
+                for a in response.json()
+                if a and "nft_addr" in a and "ve_allocated" in a and "chainID" in a
+            }
+
+        logger.info(f"veAllocate: Failed to retrieve list from {env_var} env var.")
+        return set()
+
+    def update_asset(self, asset, veAllocated):
+        """
+        Updates the field `state.allocated`  in `asset` object.
+        """
+        did = asset["id"]
+        if "stats" not in asset:
+            asset["stats"] = {}
+
+        asset["stats"]["allocated"] = veAllocated
+        logger.info(
+            f"veAllocate: updating asset {did} with state.allocated={veAllocated}."
+        )
+        try:
+            self._es_instance.update(json.dumps(asset), did)
+        except Exception as e:
+            logger.warning(f"updating ddo {did} stats.allocated attribute failed: {e}")
+
+    def update_lists(self):
+        """
+        :return: None
+        """
+        now = int(datetime.now().timestamp())
+        req_diff = int(os.getenv("VEALLOCATE_UPDATE_INTERVAL", "60")) * 60
+        if self.update_time and (now - self.update_time) < req_diff:
+            return
+
+        logger.info(
+            f"veAllocate: updating veAllocate list and setting update time to {now}."
+        )
+        self.update_time = now
+
+        ve_list = self.retrieve_new_list("VEALLOCATE_URL")
+
+        for nft, ve_allocated, chain_id in ve_list:
+            try:
+                did = make_did(nft, chain_id)
+                asset = self._es_instance.read(did)
+                self.update_asset(asset, ve_allocated)
+            except elasticsearch.exceptions.NotFoundError:
+                continue

setup.py

@@ -95,7 +95,7 @@
     url="https://github.com/oceanprotocol/aquarius",
     # fmt: off
     # bumpversion needs single quotes
-    version='4.4.2',
+    version='4.5.1',
     # fmt: on
     zip_safe=False,
 )

tests/helpers.py

@@ -63,6 +63,15 @@ def get_ddo(client, base_ddo_url, did):
         return None
 
 
+def publish_ddo(client, base_ddo_url, events_object):
+    ddo = new_ddo(test_account1, get_web3(), "dt.0")
+    did = ddo.id
+    send_create_update_tx("create", ddo, bytes([0]), test_account1)
+    events_object.process_current_blocks()
+
+    return did
+
+
 def send_create_update_tx(name, ddo, flags, account):
     provider_url = "http://172.15.0.4:8030"
     provider_address = "0xe2DD09d719Da89e5a3D0F2549c7E24566e947260"

tests/test_purgatory.py

@@ -9,13 +9,7 @@
 from requests.models import Response
 
 from aquarius.events.purgatory import Purgatory
-from tests.helpers import (
-    get_ddo,
-    get_web3,
-    new_ddo,
-    send_create_update_tx,
-    test_account1,
-)
+from tests.helpers import get_ddo, publish_ddo
 
 
 class PurgatoryForTesting(Purgatory):
@@ -32,15 +26,6 @@ def retrieve_new_list(self, env_var):
         )
 
 
-def publish_ddo(client, base_ddo_url, events_object):
-    ddo = new_ddo(test_account1, get_web3(), "dt.0")
-    did = ddo.id
-    send_create_update_tx("create", ddo, bytes([0]), test_account1)
-    events_object.process_current_blocks()
-
-    return did
-
-
 def test_purgatory_before_init(client, base_ddo_url, events_object, monkeypatch):
     monkeypatch.setenv(
         "ASSET_PURGATORY_URL",

tests/test_veallocate.py

@@ -0,0 +1,39 @@
+#
+# Copyright 2021 Ocean Protocol Foundation
+# SPDX-License-Identifier: Apache-2.0
+#
+from datetime import datetime, timedelta
+from unittest.mock import Mock, patch
+
+from freezegun import freeze_time
+from requests.models import Response
+
+from aquarius.events.ve_allocate import VeAllocate
+from tests.helpers import get_ddo, publish_ddo
+
+
+class VeAllocateForTesting(VeAllocate):
+    def __init__(self, es_instance):
+        self.current_test_asset_list = set()
+        super(VeAllocateForTesting, self).__init__(es_instance)
+
+    def retrieve_new_list(self, env_var):
+        return self.current_test_asset_list if env_var == "VEALLOCATE_URL" else None
+
+
+def test_ve_allocate_with_assets(client, base_ddo_url, events_object, monkeypatch):
+    monkeypatch.setenv(
+        "VEALLOCATE_URL",
+        "https://test-df-sql.oceandao.org/nftinfo",
+    )
+    did = publish_ddo(client, base_ddo_url, events_object)
+
+    veAllocate = VeAllocateForTesting(events_object._es_instance)
+    published_ddo = get_ddo(client, base_ddo_url, did)
+
+    veAllocate.current_test_asset_list = {
+        (published_ddo["nftAddress"], 100, published_ddo["chainId"])
+    }
+    veAllocate.update_lists()
+    published_ddo = get_ddo(client, base_ddo_url, did)
+    assert published_ddo["stats"]["allocated"] == 100