Merge pull request #5 from vsys-host/4-no-payment-notification-if-rpc-node-returns-error-while-retrieving-transaction-details

Fix payment notification if RPC node returns error

Author: yukonet

app/__init__.py

@@ -36,7 +36,6 @@ def create_app():
     from .api import metrics_blueprint
     app.register_blueprint(metrics_blueprint)
 
-    from .tasks import walletnotify_shkeeper
 
     db.init_app(app)
     with app.app_context():

app/coin.py

@@ -1,5 +1,6 @@
 from decimal import Decimal
 from flask import current_app as app
+import requests as rq
 import time
 import json
 import copy
@@ -122,6 +123,24 @@ def get_rent_amount(self) -> Decimal:
         min_balance = to_sol(min_balance)
         return min_balance
 
+    def walletnotify_shkeeper(self, symbol, txid) -> bool:
+        """Notify SHKeeper about transaction"""
+        logger.warning(f"Notifying about {symbol}/{txid}")
+        while True:
+            try:
+                r = rq.post(
+                        f'http://{config["SHKEEPER_HOST"]}/api/v1/walletnotify/{symbol}/{txid}',
+                        headers={'X-Shkeeper-Backend-Key': config['SHKEEPER_KEY']}).json()
+                if r["status"] == "success":
+                    logger.warning(f"The notification about {symbol}/{txid} was successful")
+                    return True
+                else:
+                    logger.warning(f"Failed to notify SHKeeper about {symbol}/{txid}, received response: {r}")
+                    time.sleep(5)
+            except Exception as e:
+                logger.warning(f'Shkeeper notification failed for {symbol}/{txid}: {e}')
+                time.sleep(10)
+    
     def parse_transaction(self, txid) -> list:
         "Return list of related transactions details for SHKeeper"
         if self.symbol == "SOL":

app/events.py

@@ -9,7 +9,7 @@
 
 
 def log_loop(last_checked_block, check_interval):
-    from .tasks import walletnotify_shkeeper, drain_account
+    from .tasks import drain_account
     from app import create_app
     app = create_app()
     app.app_context().push()
@@ -66,7 +66,7 @@ def check_in_parallel(block):
                                         drain_account.delay(token_dict[balance["mint"]], balance['owner'])
                     symbols_set = set(symbols)
                     for symbol in symbols_set:
-                        walletnotify_shkeeper.delay(symbol, transaction_json['transaction']['signatures'][0])
+                        coin.walletnotify_shkeeper(symbol, transaction_json['transaction']['signatures'][0])
                 return 1
             with ThreadPoolExecutor(max_workers=config['EVENTS_MAX_THREADS_NUMBER']) as executor:
                 try:

app/tasks.py

@@ -46,21 +46,6 @@ def post_payout_results(data, symbol):
             time.sleep(10)
 
 
-@celery.task()
-def walletnotify_shkeeper(symbol, txid):
-    logger.warning(f"Notifying about {symbol}/{txid}")
-    while True:
-        try:
-            r = rq.post(
-                    f'http://{config["SHKEEPER_HOST"]}/api/v1/walletnotify/{symbol}/{txid}',
-                    headers={'X-Shkeeper-Backend-Key': config['SHKEEPER_KEY']}
-                )
-            return r
-        except Exception as e:
-            logger.warning(f'Shkeeper notification failed for {symbol}/{txid}: {e}')
-            time.sleep(10)
-
-
 @celery.task()
 def refresh_balances():
     updated = 0