Add support for rari fuse pools

README.md

@@ -72,6 +72,7 @@ docker-compose -f docker-compose.yml up -d
 | AWS_SECRET_ACCESS_KEY    | The AWS secret access key used to make the oracle vote public                      | Yes      | -                                                                       |
 | STAKEWISE_SUBGRAPH_URL   | The StakeWise subgraph URL                                                         | No       | https://api.thegraph.com/subgraphs/name/stakewise/stakewise-mainnet     |
 | UNISWAP_V3_SUBGRAPH_URL  | The Uniswap V3 subgraph URL                                                        | No       | https://api.thegraph.com/subgraphs/name/stakewise/uniswap-v3-mainnet    |
+| RARI_FUSE_SUBGRAPH_URL   | Rari Capital Fuse subgraph URL                                                     | No       | https://api.thegraph.com/subgraphs/name/stakewise/rari-fuse-mainnet     |
 | ETHEREUM_SUBGRAPH_URL    | The Ethereum subgraph URL                                                          | No       | https://api.thegraph.com/subgraphs/name/stakewise/ethereum-mainnet      |
 | ORACLE_PROCESS_INTERVAL  | How long to wait before processing again (in seconds)                              | No       | 180                                                                     |
 | ETH1_CONFIRMATION_BLOCKS | The required number of ETH1 confirmation blocks used to fetch the data             | No       | 15                                                                      |

deploy/.env.example

@@ -18,6 +18,7 @@ AWS_SECRET_ACCESS_KEY=""
 # STAKEWISE_SUBGRAPH_URL=http://graph-node:8000/subgraphs/name/stakewise/stakewise
 # UNISWAP_V3_SUBGRAPH_URL=http://graph-node:8000/subgraphs/name/stakewise/uniswap-v3
 # ETHEREUM_SUBGRAPH_URL=http://graph-node:8000/subgraphs/name/stakewise/ethereum
+# RARI_FUSE_SUBGRAPH_URL=http://graph-node:8000/subgraphs/name/stakewise/rari-fuse
 
 # KEEPER
 WEB3_ENDPOINT=""

oracle/oracle/clients.py

@@ -15,6 +15,7 @@
     ETHEREUM_SUBGRAPH_URL,
     IPFS_FETCH_ENDPOINTS,
     IPFS_PIN_ENDPOINTS,
+    RARI_FUSE_SUBGRAPH_URL,
     STAKEWISE_SUBGRAPH_URL,
     UNISWAP_V3_SUBGRAPH_URL,
 )
@@ -58,6 +59,16 @@ async def execute_ethereum_gql_query(query: DocumentNode, variables: Dict) -> Di
         return await session.execute(query, variable_values=variables)
 
 
+@backoff.on_exception(backoff.expo, Exception, max_time=300, logger=gql_logger)
+async def execute_rari_fuse_pools_gql_query(
+    query: DocumentNode, variables: Dict
+) -> Dict:
+    """Executes GraphQL query."""
+    transport = AIOHTTPTransport(url=RARI_FUSE_SUBGRAPH_URL)
+    async with Client(transport=transport, execute_timeout=EXECUTE_TIMEOUT) as session:
+        return await session.execute(query, variable_values=variables)
+
+
 @backoff.on_exception(backoff.expo, Exception, max_time=900)
 async def ipfs_fetch(ipfs_hash: str) -> Union[Dict[Any, Any], List[Dict[Any, Any]]]:
     """Tries to fetch IPFS hash from different sources."""

oracle/oracle/distributor/controller.py

@@ -46,7 +46,11 @@ async def process(self, voting_params: DistributorVotingParameters) -> None:
         current_nonce = voting_params["rewards_nonce"]
 
         # skip submitting vote if too early or vote has been already submitted
-        if to_block <= last_updated_at_block or self.last_to_block == to_block:
+        if (
+            to_block <= last_updated_at_block
+            or self.last_to_block == to_block
+            or from_block >= to_block
+        ):
             return
 
         logger.info(
@@ -71,6 +75,7 @@ async def process(self, voting_params: DistributorVotingParameters) -> None:
         disabled_stakers_distributions = (
             await get_disabled_stakers_reward_eth_distributions(
                 distributor_reward=voting_params["distributor_reward"],
+                from_block=from_block,
                 to_block=to_block,
             )
         )
@@ -102,7 +107,8 @@ async def process(self, voting_params: DistributorVotingParameters) -> None:
         for dist in all_distributions:
             distributor_rewards = DistributorRewards(
                 uniswap_v3_pools=uniswap_v3_pools,
-                block_number=dist["block_number"],
+                from_block=dist["from_block"],
+                to_block=dist["to_block"],
                 reward_token=dist["reward_token"],
                 uni_v3_token=dist["uni_v3_token"],
                 swise_holders=swise_holders,
@@ -131,7 +137,8 @@ async def process(self, voting_params: DistributorVotingParameters) -> None:
         swise_holders_rewards = await DistributorRewards(
             uniswap_v3_pools=uniswap_v3_pools,
             swise_holders=swise_holders,
-            block_number=to_block,
+            from_block=from_block,
+            to_block=to_block,
             uni_v3_token=SWISE_TOKEN_CONTRACT_ADDRESS,
             reward_token=REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
         ).get_rewards(SWISE_TOKEN_CONTRACT_ADDRESS, left_reward)

oracle/oracle/distributor/eth1.py

@@ -87,7 +87,7 @@ async def get_periodic_allocations(
 
 @backoff.on_exception(backoff.expo, Exception, max_time=900)
 async def get_disabled_stakers_reward_eth_distributions(
-    distributor_reward: Wei, to_block: BlockNumber
+    distributor_reward: Wei, from_block: BlockNumber, to_block: BlockNumber
 ) -> Distributions:
     """Fetches disabled stakers reward ETH distributions based on their staked ETH balances."""
     if distributor_reward <= 0:
@@ -149,7 +149,8 @@ async def get_disabled_stakers_reward_eth_distributions(
 
         distribution = Distribution(
             contract=staker_address,
-            block_number=to_block,
+            from_block=from_block,
+            to_block=to_block,
             uni_v3_token=STAKED_ETH_TOKEN_CONTRACT_ADDRESS,
             reward_token=REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
             reward=reward,

oracle/oracle/distributor/rari.py

@@ -0,0 +1,70 @@
+from typing import Dict
+
+import backoff
+from ens.constants import EMPTY_ADDR_HEX
+from eth_typing import BlockNumber, ChecksumAddress
+from web3 import Web3
+
+from oracle.oracle.clients import execute_rari_fuse_pools_gql_query
+
+from ..graphql_queries import RARI_FUSE_POOLS_CTOKENS_QUERY
+from .types import Balances
+
+
+@backoff.on_exception(backoff.expo, Exception, max_time=900)
+async def get_rari_fuse_liquidity_points(
+    ctoken_address: ChecksumAddress, from_block: BlockNumber, to_block: BlockNumber
+) -> Balances:
+    """Fetches Rari Fuse pools."""
+    lowered_ctoken_address = ctoken_address.lower()
+    last_id = ""
+    result: Dict = await execute_rari_fuse_pools_gql_query(
+        query=RARI_FUSE_POOLS_CTOKENS_QUERY,
+        variables=dict(
+            ctoken_address=lowered_ctoken_address,
+            block_number=to_block,
+            last_id=last_id,
+        ),
+    )
+    positions_chunk = result.get("accountCTokens", [])
+    positions = positions_chunk
+
+    # accumulate chunks of positions
+    while len(positions_chunk) >= 1000:
+        last_id = positions_chunk[-1]["id"]
+        result: Dict = await execute_rari_fuse_pools_gql_query(
+            query=RARI_FUSE_POOLS_CTOKENS_QUERY,
+            variables=dict(
+                ctoken_address=lowered_ctoken_address,
+                block_number=to_block,
+                last_id=last_id,
+            ),
+        )
+        positions_chunk = result.get("accountCTokens", [])
+        positions.extend(positions_chunk)
+
+    # process fuse pools balances
+    points: Dict[ChecksumAddress, int] = {}
+    total_points = 0
+    for position in positions:
+        account = Web3.toChecksumAddress(position["account"])
+        if account == EMPTY_ADDR_HEX:
+            continue
+
+        principal = int(position["cTokenBalance"])
+        prev_account_points = int(position["distributorPoints"])
+        updated_at_block = BlockNumber(int(position["updatedAtBlock"]))
+        if from_block > updated_at_block:
+            updated_at_block = from_block
+            prev_account_points = 0
+
+        account_points = prev_account_points + (
+            principal * (to_block - updated_at_block)
+        )
+        if account_points <= 0:
+            continue
+
+        points[account] = points.get(account, 0) + account_points
+        total_points += account_points
+
+    return Balances(total_supply=total_points, balances=points)

oracle/oracle/distributor/rewards.py

@@ -7,11 +7,13 @@
 
 from oracle.oracle.settings import (
     DISTRIBUTOR_FALLBACK_ADDRESS,
+    RARI_FUSE_POOL_ADDRESSES,
     REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
     STAKED_ETH_TOKEN_CONTRACT_ADDRESS,
     SWISE_TOKEN_CONTRACT_ADDRESS,
 )
 
+from .rari import get_rari_fuse_liquidity_points
 from .types import Balances, Rewards, UniswapV3Pools
 from .uniswap_v3 import (
     get_uniswap_v3_liquidity_points,
@@ -26,7 +28,8 @@ class DistributorRewards(object):
     def __init__(
         self,
         uniswap_v3_pools: UniswapV3Pools,
-        block_number: BlockNumber,
+        from_block: BlockNumber,
+        to_block: BlockNumber,
         reward_token: ChecksumAddress,
         uni_v3_token: ChecksumAddress,
         swise_holders: Balances,
@@ -37,7 +40,8 @@ def __init__(
         self.uni_v3_pools = self.uni_v3_swise_pools.union(
             self.uni_v3_staked_eth_pools
         ).union(self.uni_v3_reward_eth_pools)
-        self.block_number = block_number
+        self.from_block = from_block
+        self.to_block = to_block
         self.uni_v3_token = uni_v3_token
         self.reward_token = reward_token
         self.swise_holders = swise_holders
@@ -47,6 +51,7 @@ def is_supported_contract(self, contract_address: ChecksumAddress) -> bool:
         return (
             contract_address in self.uni_v3_pools
             or contract_address == SWISE_TOKEN_CONTRACT_ADDRESS
+            or contract_address in RARI_FUSE_POOL_ADDRESSES
         )
 
     @staticmethod
@@ -110,7 +115,7 @@ async def get_balances(self, contract_address: ChecksumAddress) -> Balances:
             return await get_uniswap_v3_single_token_balances(
                 pool_address=contract_address,
                 token=STAKED_ETH_TOKEN_CONTRACT_ADDRESS,
-                block_number=self.block_number,
+                block_number=self.to_block,
             )
         elif (
             self.uni_v3_token == REWARD_ETH_TOKEN_CONTRACT_ADDRESS
@@ -120,7 +125,7 @@ async def get_balances(self, contract_address: ChecksumAddress) -> Balances:
             return await get_uniswap_v3_single_token_balances(
                 pool_address=contract_address,
                 token=REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
-                block_number=self.block_number,
+                block_number=self.to_block,
             )
         elif (
             self.uni_v3_token == SWISE_TOKEN_CONTRACT_ADDRESS
@@ -130,7 +135,7 @@ async def get_balances(self, contract_address: ChecksumAddress) -> Balances:
             return await get_uniswap_v3_single_token_balances(
                 pool_address=contract_address,
                 token=SWISE_TOKEN_CONTRACT_ADDRESS,
-                block_number=self.block_number,
+                block_number=self.to_block,
             )
         elif (
             self.uni_v3_token == EMPTY_ADDR_HEX
@@ -143,15 +148,24 @@ async def get_balances(self, contract_address: ChecksumAddress) -> Balances:
                 tick_lower=-887220,
                 tick_upper=887220,
                 pool_address=contract_address,
-                block_number=self.block_number,
+                block_number=self.to_block,
             )
         elif contract_address in self.uni_v3_pools:
             logger.info(
                 f"Fetching Uniswap V3 liquidity points: pool={contract_address}"
             )
             return await get_uniswap_v3_liquidity_points(
                 pool_address=contract_address,
-                block_number=self.block_number,
+                block_number=self.to_block,
+            )
+        elif contract_address in RARI_FUSE_POOL_ADDRESSES:
+            logger.info(
+                f"Fetching Rari Fuse Pool liquidity points: pool={contract_address}"
+            )
+            return await get_rari_fuse_liquidity_points(
+                ctoken_address=contract_address,
+                from_block=self.from_block,
+                to_block=self.to_block,
             )
         elif contract_address == SWISE_TOKEN_CONTRACT_ADDRESS:
             logger.info("Distributing rewards to SWISE holders")

oracle/oracle/distributor/types.py

@@ -17,7 +17,8 @@ class DistributorVotingParameters(TypedDict):
 
 class Distribution(TypedDict):
     contract: ChecksumAddress
-    block_number: BlockNumber
+    from_block: BlockNumber
+    to_block: BlockNumber
     uni_v3_token: ChecksumAddress
     reward_token: ChecksumAddress
     reward: int

oracle/oracle/distributor/uniswap_v3.py

@@ -125,24 +125,26 @@ async def get_uniswap_v3_distributions(
                     if reward > 0:
                         distribution = Distribution(
                             contract=pool_address,
-                            block_number=BlockNumber(start + interval),
+                            from_block=start,
+                            to_block=BlockNumber(start + interval),
                             reward_token=allocation["reward_token"],
                             reward=reward,
                             uni_v3_token=EMPTY_ADDR_HEX,
                         )
                         distributions.append(distribution)
                     break
 
-                start += BLOCKS_INTERVAL
                 if interval_reward > 0:
                     distribution = Distribution(
                         contract=pool_address,
-                        block_number=start,
+                        from_block=start,
+                        to_block=BlockNumber(start + BLOCKS_INTERVAL),
                         reward_token=allocation["reward_token"],
                         reward=interval_reward,
                         uni_v3_token=EMPTY_ADDR_HEX,
                     )
                     distributions.append(distribution)
+                start += BLOCKS_INTERVAL
 
     return distributions
 

oracle/oracle/graphql_queries.py

@@ -297,6 +297,30 @@
 """
 )
 
+RARI_FUSE_POOLS_CTOKENS_QUERY = gql(
+    """
+    query getCTokens(
+      $block_number: Int
+      $ctoken_address: Bytes
+      $last_id: ID
+    ) {
+      accountCTokens(
+        block: { number: $block_number }
+        where: { ctoken: $ctoken_address, id_gt: $last_id }
+        first: 1000
+        orderBy: id
+        orderDirection: asc
+      ) {
+        id
+        account
+        cTokenBalance
+        distributorPoints
+        updatedAtBlock
+      }
+    }
+"""
+)
+
 DISTRIBUTOR_CLAIMED_ACCOUNTS_QUERY = gql(
     """
     query getDistributorClaims($merkle_root: Bytes, $last_id: ID) {

oracle/oracle/settings.py

@@ -63,6 +63,10 @@
     DISTRIBUTOR_FALLBACK_ADDRESS = Web3.toChecksumAddress(
         "0x144a98cb1CdBb23610501fE6108858D9B7D24934"
     )
+    RARI_FUSE_POOL_ADDRESSES = [
+        Web3.toChecksumAddress("0x18F49849D20Bc04059FE9d775df9a38Cd1f5eC9F"),
+        Web3.toChecksumAddress("0x83d534Ab1d4002249B0E6d22410b62CF31978Ca2"),
+    ]
     WITHDRAWAL_CREDENTIALS: HexStr = HexStr(
         "0x0100000000000000000000002296e122c1a20fca3cac3371357bdad3be0df079"
     )
@@ -78,6 +82,10 @@
         "ETHEREUM_SUBGRAPH_URL",
         default="https://api.thegraph.com/subgraphs/name/stakewise/ethereum-mainnet",
     )
+    RARI_FUSE_SUBGRAPH_URL = config(
+        "RARI_FUSE_SUBGRAPH_URL",
+        default="https://api.thegraph.com/subgraphs/name/stakewise/rari-fuse-mainnet",
+    )
 elif NETWORK == GOERLI:
     SYNC_PERIOD = timedelta(hours=1)
     SWISE_TOKEN_CONTRACT_ADDRESS = Web3.toChecksumAddress(
@@ -92,6 +100,7 @@
     DISTRIBUTOR_FALLBACK_ADDRESS = Web3.toChecksumAddress(
         "0x1867c96601bc5fE24F685d112314B8F3Fe228D5A"
     )
+    RARI_FUSE_POOL_ADDRESSES = []
     WITHDRAWAL_CREDENTIALS: HexStr = HexStr(
         "0x010000000000000000000000040f15c6b5bfc5f324ecab5864c38d4e1eef4218"
     )
@@ -107,3 +116,4 @@
         "ETHEREUM_SUBGRAPH_URL",
         default="https://api.thegraph.com/subgraphs/name/stakewise/ethereum-goerli",
     )
+    RARI_FUSE_SUBGRAPH_URL = ""