Fix invalid webhook handling

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/src/crons/webhooks/_common.py

@@ -18,6 +18,7 @@ def handle_webhook(
     session: Session,
     webhook: Webhook,
     *,
+    queue: webhook_service.OracleWebhookQueue,
     on_fail: Callable[[Session, Webhook, Exception], None] = lambda _s, _w, _e: None,
 ):
     logger.debug(
@@ -39,9 +40,9 @@ def handle_webhook(
             savepoint.rollback()
             raise
         finally:
-            webhook_service.outbox.handle_webhook_fail(session, webhook.id)
+            queue.handle_webhook_fail(session, webhook.id)
     else:
-        webhook_service.outbox.handle_webhook_success(session, webhook.id)
+        queue.handle_webhook_success(session, webhook.id)
         logger.debug("Webhook handled successfully")
 
 
@@ -80,6 +81,6 @@ def process_outgoing_webhooks(
         for_update=ForUpdateParams(skip_locked=True),
     )
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook):
+        with handle_webhook(logger, session, webhook, queue=webhook_service.outbox):
             webhook_url = url_getter(webhook.chain_id, webhook.escrow_address)
             _send_webhook(webhook_url, webhook, with_timestamp=with_timestamp)

packages/examples/cvat/exchange-oracle/src/crons/webhooks/job_launcher.py

@@ -53,7 +53,9 @@ def process_incoming_job_launcher_webhooks(logger: logging.Logger, session: Sess
     )
 
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook, on_fail=handle_failure):
+        with handle_webhook(
+            logger, session, webhook, on_fail=handle_failure, queue=oracle_db_service.inbox
+        ):
             handle_job_launcher_event(webhook, db_session=session, logger=logger)
 
 

packages/examples/cvat/exchange-oracle/src/crons/webhooks/recording_oracle.py

@@ -35,7 +35,7 @@ def process_incoming_recording_oracle_webhooks(logger: logging.Logger, session:
     )
 
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook):
+        with handle_webhook(logger, session, webhook, queue=oracle_db_service.inbox):
             handle_recording_oracle_event(webhook, db_session=session, logger=logger)
 
 

packages/examples/cvat/exchange-oracle/src/crons/webhooks/reputation_oracle.py

@@ -23,7 +23,7 @@ def process_incoming_reputation_oracle_webhooks(logger: logging.Logger, session:
         for_update=ForUpdateParams(skip_locked=True),
     )
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook):
+        with handle_webhook(logger, session, webhook, queue=oracle_db_service.inbox):
             match webhook.event_type:
                 case ReputationOracleEventTypes.escrow_completed:
                     projects = db_service.get_projects_by_escrow_address(

packages/examples/cvat/recording-oracle/src/crons/_utils.py

@@ -69,7 +69,13 @@ def wrapper():
 
 
 @contextmanager
-def handle_webhook(logger: logging.Logger, session: Session, webhook: Webhook):
+def handle_webhook(
+    logger: logging.Logger,
+    session: Session,
+    webhook: Webhook,
+    *,
+    queue: webhook_service.OracleWebhookQueue,
+):
     savepoint = session.begin_nested()
     logger.debug(
         "Processing webhook "
@@ -83,9 +89,9 @@ def handle_webhook(logger: logging.Logger, session: Session, webhook: Webhook):
         # TODO: should we rollback on any errors or just on database errors?
         savepoint.rollback()
         logger.exception(f"Webhook {webhook.id} sending failed: {e}")
-        webhook_service.outbox.handle_webhook_fail(session, webhook.id)
+        queue.handle_webhook_fail(session, webhook.id)
     else:
-        webhook_service.outbox.handle_webhook_success(session, webhook.id)
+        queue.handle_webhook_success(session, webhook.id)
         logger.debug("Webhook handled successfully")
 
 
@@ -124,6 +130,6 @@ def process_outgoing_webhooks(
         for_update=ForUpdateParams(skip_locked=True),
     )
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook):
+        with handle_webhook(logger, session, webhook, queue=oracle_db_service.outbox):
             webhook_url = url_getter(webhook.chain_id, webhook.escrow_address)
             send_webhook(webhook_url, webhook, with_timestamp=with_timestamp)

packages/examples/cvat/recording-oracle/src/crons/process_exchange_oracle_webhooks.py

@@ -26,7 +26,7 @@ def process_incoming_exchange_oracle_webhooks(logger: logging.Logger, session: S
     )
 
     for webhook in webhooks:
-        with handle_webhook(logger, session, webhook):
+        with handle_webhook(logger, session, webhook, queue=oracle_db_service.inbox):
             handle_exchange_oracle_event(webhook, db_session=session)