[CVAT][Exchange Oracle] Data cleanup

packages/examples/cvat/exchange-oracle/pyproject.toml

@@ -31,6 +31,7 @@ hexbytes = ">=1.2.0" # required for to_0x_hex() function
 [tool.poetry.group.dev.dependencies]
 pre-commit = "^3.0.4"
 ruff = "^0.6.0"
+pytest-mock = "^3.14.0"
 
 [tool.ruff]
 line-length = 100
@@ -99,7 +100,6 @@ ignore = [
     "ERA001",  # Found commented-out code
     "N801",    # Class name should use CapWords convention
     "PLR0915", # Too many statements
-    "F401",    # Imported but unused
     "PLR2004", # Magic value used in comparison, consider replacing with a constant variable
     "ANN002",  # Missing type annotation for `*args`
     "TRY300",  # Consider moving this statement to an `else` block
@@ -130,6 +130,7 @@ ignore = [
 ]
 # alembic is not a package in a traditional sense, so putting __init__.py there doesn't make sense
 "alembic/*" = ["INP001"]
+"__init__.py" = ["F401"]
 
 [tool.ruff.lint.pep8-naming]
 classmethod-decorators = [

packages/examples/cvat/exchange-oracle/src/.env.template

@@ -32,6 +32,8 @@ PROCESS_JOB_LAUNCHER_WEBHOOKS_INT=
 PROCESS_JOB_LAUNCHER_WEBHOOKS_CHUNK_SIZE=
 PROCESS_RECORDING_ORACLE_WEBHOOKS_INT=
 PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE=
+PROCESS_REPUTATION_ORACLE_WEBHOOKS_CHUNK_SIZE=
+PROCESS_REPUTATION_ORACLE_WEBHOOKS_INT=
 TRACK_COMPLETED_PROJECTS_INT=
 TRACK_COMPLETED_PROJECTS_CHUNK_SIZE=
 TRACK_COMPLETED_TASKS_INT=
@@ -90,6 +92,7 @@ HUMAN_APP_SIGNATURE=
 LOCALHOST_RECORDING_ORACLE_ADDRESS=
 LOCALHOST_RECORDING_ORACLE_URL=
 LOCALHOST_JOB_LAUNCHER_URL=
+LOCALHOST_REPUTATION_ORACLE_URL=
 
 # Encryption
 PGP_PRIVATE_KEY=

packages/examples/cvat/exchange-oracle/src/chain/escrow.py

@@ -6,6 +6,7 @@
 from human_protocol_sdk.storage import StorageUtils
 
 from src.core.config import Config
+from src.core.types import OracleWebhookTypes
 
 
 def get_escrow(chain_id: int, escrow_address: str) -> EscrowData:
@@ -56,12 +57,16 @@ def get_escrow_manifest(chain_id: int, escrow_address: str) -> dict:
     return json.loads(manifest_content)
 
 
-def get_job_launcher_address(chain_id: int, escrow_address: str) -> str:
-    return get_escrow(chain_id, escrow_address).launcher
-
-
-def get_recording_oracle_address(chain_id: int, escrow_address: str) -> str:
-    if address := Config.localhost.recording_oracle_address:
-        return address
-
-    return get_escrow(chain_id, escrow_address).recording_oracle
+def get_available_webhook_types(
+    chain_id: int, escrow_address: str
+) -> dict[str, OracleWebhookTypes]:
+    escrow = get_escrow(chain_id, escrow_address)
+    return {
+        escrow.launcher.lower(): OracleWebhookTypes.job_launcher,
+        (
+            Config.localhost.recording_oracle_address or escrow.recording_oracle
+        ).lower(): OracleWebhookTypes.recording_oracle,
+        (
+            Config.localhost.reputation_oracle_url or escrow.reputation_oracle
+        ).lower(): OracleWebhookTypes.reputation_oracle,
+    }

packages/examples/cvat/exchange-oracle/src/chain/kvstore.py

@@ -16,6 +16,15 @@ def get_recording_oracle_url(chain_id: int, escrow_address: str) -> str:
     return OperatorUtils.get_leader(ChainId(chain_id), escrow.recording_oracle).webhook_url
 
 
+def get_reputation_oracle_url(chain_id: int, escrow_address: str) -> str:
+    if url := Config.localhost.recording_oracle_url:
+        return url
+
+    escrow = get_escrow(chain_id, escrow_address)
+
+    return OperatorUtils.get_leader(ChainId(chain_id), escrow.recording_oracle).webhook_url
+
+
 def get_job_launcher_url(chain_id: int, escrow_address: str) -> str:
     if url := Config.localhost.job_launcher_url:
         return url

packages/examples/cvat/exchange-oracle/src/core/config.py

@@ -83,18 +83,25 @@ class LocalhostConfig(_NetworkConfig):
 
     recording_oracle_address = os.environ.get("LOCALHOST_RECORDING_ORACLE_ADDRESS")
     recording_oracle_url = os.environ.get("LOCALHOST_RECORDING_ORACLE_URL")
+    reputation_oracle_url = os.environ.get("LOCALHOST_REPUTATION_ORACLE_URL")
 
 
 class CronConfig:
     process_job_launcher_webhooks_int = int(os.environ.get("PROCESS_JOB_LAUNCHER_WEBHOOKS_INT", 30))
-    process_job_launcher_webhooks_chunk_size = os.environ.get(
-        "PROCESS_JOB_LAUNCHER_WEBHOOKS_CHUNK_SIZE", 5
+    process_job_launcher_webhooks_chunk_size = int(
+        os.environ.get("PROCESS_JOB_LAUNCHER_WEBHOOKS_CHUNK_SIZE", 5)
     )
     process_recording_oracle_webhooks_int = int(
         os.environ.get("PROCESS_RECORDING_ORACLE_WEBHOOKS_INT", 30)
     )
-    process_recording_oracle_webhooks_chunk_size = os.environ.get(
-        "PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE", 5
+    process_recording_oracle_webhooks_chunk_size = int(
+        os.environ.get("PROCESS_RECORDING_ORACLE_WEBHOOKS_CHUNK_SIZE", 5)
+    )
+    process_reputation_oracle_webhooks_chunk_size = int(
+        os.environ.get("PROCESS_REPUTATION_ORACLE_WEBHOOKS_CHUNK_SIZE", 5)
+    )
+    process_reputation_oracle_webhooks_int = int(
+        os.environ.get("PROCESS_REPUTATION_ORACLE_WEBHOOKS_INT", 5)
     )
     track_completed_projects_int = int(os.environ.get("TRACK_COMPLETED_PROJECTS_INT", 30))
     track_completed_projects_chunk_size = os.environ.get("TRACK_COMPLETED_PROJECTS_CHUNK_SIZE", 5)

packages/examples/cvat/exchange-oracle/src/core/oracle_events.py

@@ -1,12 +1,11 @@
-from typing import Union
-
 from pydantic import BaseModel
 
 from src.core.types import (
     ExchangeOracleEventTypes,
     JobLauncherEventTypes,
     OracleWebhookTypes,
     RecordingOracleEventTypes,
+    ReputationOracleEventTypes,
 )
 
 EventTypeTag = ExchangeOracleEventTypes | JobLauncherEventTypes | RecordingOracleEventTypes
@@ -44,13 +43,23 @@ class ExchangeOracleEvent_TaskFinished(OracleEvent):
     pass  # escrow is enough for now
 
 
+class ExchangeOracleEvent_EscrowCleaned(OracleEvent):
+    pass
+
+
+class ReputationOracleEvent_EscrowCompleted(OracleEvent):
+    pass
+
+
 _event_type_map = {
     JobLauncherEventTypes.escrow_created: JobLauncherEvent_EscrowCreated,
     JobLauncherEventTypes.escrow_canceled: JobLauncherEvent_EscrowCanceled,
     RecordingOracleEventTypes.task_completed: RecordingOracleEvent_TaskCompleted,
     RecordingOracleEventTypes.task_rejected: RecordingOracleEvent_TaskRejected,
     ExchangeOracleEventTypes.task_creation_failed: ExchangeOracleEvent_TaskCreationFailed,
     ExchangeOracleEventTypes.task_finished: ExchangeOracleEvent_TaskFinished,
+    ExchangeOracleEventTypes.escrow_cleaned: ExchangeOracleEvent_EscrowCleaned,
+    ReputationOracleEventTypes.escrow_completed: ReputationOracleEvent_EscrowCompleted,
 }
 
 
@@ -83,6 +92,7 @@ def parse_event(
         OracleWebhookTypes.job_launcher: JobLauncherEventTypes,
         OracleWebhookTypes.recording_oracle: RecordingOracleEventTypes,
         OracleWebhookTypes.exchange_oracle: ExchangeOracleEventTypes,
+        OracleWebhookTypes.reputation_oracle: ReputationOracleEventTypes,
     }
 
     sender_events = sender_events_mapping.get(sender)

packages/examples/cvat/exchange-oracle/src/core/storage.py

@@ -2,9 +2,17 @@
 from src.core.types import Networks
 
 
+def compose_data_bucket_prefix(escrow_address: str, chain_id: Networks):
+    return f"{escrow_address}@{chain_id}"
+
+
+def compose_results_bucket_prefix(escrow_address: str, chain_id: Networks):
+    return f"{escrow_address}@{chain_id}{Config.storage_config.results_dir_suffix}"
+
+
 def compose_data_bucket_filename(escrow_address: str, chain_id: Networks, filename: str) -> str:
-    return f"{escrow_address}@{chain_id}/{filename}"
+    return f"{compose_data_bucket_prefix(escrow_address, chain_id)}/{filename}"
 
 
 def compose_results_bucket_filename(escrow_address: str, chain_id: Networks, filename: str) -> str:
-    return f"{escrow_address}@{chain_id}{Config.storage_config.results_dir_suffix}/{filename}"
+    return f"{compose_results_bucket_prefix(escrow_address, chain_id)}/{filename}"

packages/examples/cvat/exchange-oracle/src/core/types.py

@@ -23,6 +23,7 @@ class ProjectStatuses(str, Enum, metaclass=BetterEnumMeta):
     validation = "validation"
     canceled = "canceled"
     recorded = "recorded"
+    deleted = "deleted"
 
 
 class TaskStatuses(str, Enum, metaclass=BetterEnumMeta):
@@ -55,11 +56,13 @@ class OracleWebhookTypes(str, Enum, metaclass=BetterEnumMeta):
     exchange_oracle = "exchange_oracle"
     job_launcher = "job_launcher"
     recording_oracle = "recording_oracle"
+    reputation_oracle = "reputation_oracle"
 
 
 class ExchangeOracleEventTypes(str, Enum, metaclass=BetterEnumMeta):
     task_creation_failed = "task_creation_failed"
     task_finished = "task_finished"
+    escrow_cleaned = "escrow_cleaned"
 
 
 class JobLauncherEventTypes(str, Enum, metaclass=BetterEnumMeta):
@@ -72,6 +75,11 @@ class RecordingOracleEventTypes(str, Enum, metaclass=BetterEnumMeta):
     task_rejected = "task_rejected"
 
 
+class ReputationOracleEventTypes(str, Enum, metaclass=BetterEnumMeta):
+    # TODO: rename to ReputationOracleEventType
+    escrow_completed = "escrow_completed"
+
+
 class OracleWebhookStatuses(str, Enum, metaclass=BetterEnumMeta):
     pending = "pending"
     completed = "completed"