Merge branch 'master' into rube/heartbeat-block

Author: rube-de

.env.example

@@ -5,7 +5,7 @@ ROFL_ADAPTER_ADDRESS=YOUR_ROFL_ADAPTER_ADDRESS  # ROFLAdapter contract address o
 REQUEST_TIMEOUT=30  # HTTP request timeout in seconds
 RETRY_COUNT=3  # Number of retry attempts for failed operations
 
-# Oracle mode: event_listener, push, or watcher
+# Oracle mode: event_listener, push, watcher, or token_watcher
 ORACLE_MODE=event_listener
 
 # Logging configuration
@@ -28,6 +28,13 @@ WATCHER_BATCH_SIZE=50  # Max blocks to scan per iteration
 # LOOKBACK_BLOCKS=100  # Also used in watcher mode
 ENABLE_INTERNAL_TX_DETECTION=false  # Enable internal transaction detection (requires archive node with debug API)
 
+# Token Watcher Mode Configuration (ORACLE_MODE=token_watcher)
+TOKEN_ADDRESSES=0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48  # Comma-separated ERC20 token contract addresses (e.g., USDC, USDT)
+RECIPIENT_ADDRESSES=0xYourAddress1,0xYourAddress2  # Comma-separated recipient addresses to watch for
+SCAN_INTERVAL=5  # Seconds between scans (default: 5 for fast detection)
+# Note: On first startup, begins monitoring from current block (no historical scan)
+# Note: Scans all new blocks since last check (no batch size limit)
+
 # Local mode configuration (for testing without ROFL)
 # Only required when running with --local flag or compose.local.yaml
 LOCAL_PRIVATE_KEY=0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef

compose.local.yaml

@@ -6,7 +6,7 @@ services:
     image: ghcr.io/rube-de/rofl-header-oracle:latest
     platform: linux/amd64
     container_name: rofl_header_oracle
-    entrypoint: /bin/sh -c 'uv run python -m rofl_oracle --local --log-level DEBUG'
+    entrypoint: /bin/sh -c 'uv run python -m rofl_oracle --local'
     ports:
       - "8080:8080"  # Health check endpoint
     environment:
@@ -20,12 +20,12 @@ services:
       REQUEST_TIMEOUT: ${REQUEST_TIMEOUT:-30}  # HTTP request timeout in seconds
       RETRY_COUNT: ${RETRY_COUNT:-3}  # Number of retry attempts
 
-      # Oracle mode: event_listener, push, or watcher
+      # Oracle mode: event_listener, push, watcher, or token_watcher
       ORACLE_MODE: ${ORACLE_MODE:-event_listener}
 
       # Logging configuration
-      LOG_LEVEL: ${LOG_LEVEL:-DEBUG}  # DEBUG mode for local testing
-      JSON_LOGS: ${JSON_LOGS:-false}  # Enable JSON-formatted structured logging
+      LOG_LEVEL: ${LOG_LEVEL:-INFO}
+      JSON_LOGS: ${JSON_LOGS:-false}
 
       # Event Listener Mode Configuration (ORACLE_MODE=event_listener)
       SOURCE_CONTRACT_ADDRESS: ${SOURCE_CONTRACT_ADDRESS:-}  # Required for event_listener mode
@@ -42,8 +42,12 @@ services:
       WATCHER_BATCH_SIZE: ${WATCHER_BATCH_SIZE:-50}  # Max blocks to scan per iteration
       ENABLE_INTERNAL_TX_DETECTION: ${ENABLE_INTERNAL_TX_DETECTION:-false}  # Requires archive node with debug API
 
+      # Token Watcher Mode Configuration (ORACLE_MODE=token_watcher)
+      TOKEN_ADDRESSES: ${TOKEN_ADDRESSES:-}  # Required for token_watcher mode - comma-separated token contract addresses
+      RECIPIENT_ADDRESSES: ${RECIPIENT_ADDRESSES:-}  # Required for token_watcher mode - comma-separated recipient addresses
+
       # Local mode configuration (for testing without ROFL)
       LOCAL_PRIVATE_KEY: ${LOCAL_PRIVATE_KEY:-}  # Required in local mode - private key for testing
-    
+
     restart: on-failure
     # Note: No ROFL socket volume mount needed in local mode

compose.yaml

@@ -40,7 +40,11 @@ services:
       SCAN_INTERVAL: ${SCAN_INTERVAL}  # Seconds between scanning for interactions
       WATCHER_BATCH_SIZE: ${WATCHER_BATCH_SIZE}  # Max blocks to scan per iteration
       ENABLE_INTERNAL_TX_DETECTION: ${ENABLE_INTERNAL_TX_DETECTION:-false}  # Requires archive node with debug API
-    
+
+      # Token Watcher Mode Configuration (ORACLE_MODE=token_watcher)
+      TOKEN_ADDRESSES: ${TOKEN_ADDRESSES:-}  # Required for token_watcher mode - comma-separated token contract addresses
+      RECIPIENT_ADDRESSES: ${RECIPIENT_ADDRESSES:-}  # Required for token_watcher mode - comma-separated recipient addresses
+
     restart: on-failure
     volumes:
       - /run/rofl-appd.sock:/run/rofl-appd.sock

rofl_oracle/config.py

@@ -20,9 +20,10 @@
 class OracleMode(Enum):
     """Oracle operating modes."""
 
-    EVENT_LISTENER = "event_listener"  # Listen for BlockHeaderRequested events
-    PUSH = "push"  # Push latest block headers periodically
-    WATCHER = "watcher"  # Watch specific addresses for interactions
+    EVENT_LISTENER = "event_listener"
+    PUSH = "push"
+    WATCHER = "watcher"
+    TOKEN_WATCHER = "token_watcher"
 
 
 @dataclass(frozen=True, slots=True)
@@ -202,6 +203,71 @@ def __post_init__(self) -> None:
             object.__setattr__(self, "watch_addresses", checksummed_addresses)
 
 
+@dataclass(frozen=True, slots=True)
+class TokenWatcherModeConfig:
+    """Configuration specific to token watcher mode.
+
+    This mode watches for ERC20 token transfers to specific recipient addresses
+    and pushes block headers when such transfers are detected.
+
+    :cvar token_addresses: List of ERC20 token contract addresses to monitor (env: TOKEN_ADDRESSES, comma-separated)
+    :cvar recipient_addresses: List of recipient addresses to watch for (env: RECIPIENT_ADDRESSES, comma-separated)
+    :cvar scan_interval: Seconds between scanning for token transfers (env: SCAN_INTERVAL)
+    :cvar max_blocks_per_scan: Maximum number of blocks to scan per iteration (env: MAX_BLOCKS_PER_SCAN, default: 10)
+    """
+
+    token_addresses: list[str]
+    recipient_addresses: list[str]
+    scan_interval: int
+    max_blocks_per_scan: int = 10
+
+    def __post_init__(self) -> None:
+        """Validate token watcher configuration."""
+        if self.scan_interval <= 0:
+            raise ValueError(
+                f"Scan interval must be positive, got {self.scan_interval}"
+            )
+        if self.scan_interval > 300:
+            raise ValueError(
+                f"Scan interval too long (max 300s), got {self.scan_interval}"
+            )
+
+        if self.max_blocks_per_scan <= 0:
+            raise ValueError(
+                f"Max blocks per scan must be positive, got {self.max_blocks_per_scan}"
+            )
+        if self.max_blocks_per_scan > 10000:
+            raise ValueError(
+                f"Max blocks per scan too high (max 10000), got {self.max_blocks_per_scan}"
+            )
+
+        if not self.token_addresses or len(self.token_addresses) == 0:
+            raise ValueError(
+                "Token watcher mode requires at least one token address"
+            )
+
+        if not self.recipient_addresses or len(self.recipient_addresses) == 0:
+            raise ValueError(
+                "Token watcher mode requires at least one recipient address"
+            )
+
+        # Validate and checksum token addresses
+        checksummed_tokens = []
+        for addr in self.token_addresses:
+            if not Web3.is_address(addr):
+                raise ValueError(f"Invalid token address: {addr}")
+            checksummed_tokens.append(Web3.to_checksum_address(addr))
+        object.__setattr__(self, "token_addresses", checksummed_tokens)
+
+        # Validate and checksum recipient addresses
+        checksummed_recipients = []
+        for addr in self.recipient_addresses:
+            if not Web3.is_address(addr):
+                raise ValueError(f"Invalid recipient address: {addr}")
+            checksummed_recipients.append(Web3.to_checksum_address(addr))
+        object.__setattr__(self, "recipient_addresses", checksummed_recipients)
+
+
 @dataclass(frozen=True, slots=True)
 class CommonConfig:
     """Configuration common to all oracle modes.
@@ -294,14 +360,19 @@ class OracleConfig:
 
     :cvar common_config: Common configuration shared across all modes
     :cvar oracle_mode: The operating mode of the oracle
-    :cvar mode_config: Mode-specific configuration (EventListenerModeConfig, PushModeConfig, or WatcherModeConfig)
+    :cvar mode_config: Mode-specific configuration (EventListenerModeConfig, PushModeConfig, WatcherModeConfig, or TokenWatcherModeConfig)
     :cvar local_mode: Whether running in local mode (for testing)
     :cvar local_private_key: Private key for local mode (optional)
     """
 
     common_config: CommonConfig
     oracle_mode: OracleMode
-    mode_config: EventListenerModeConfig | PushModeConfig | WatcherModeConfig
+    mode_config: (
+        EventListenerModeConfig
+        | PushModeConfig
+        | WatcherModeConfig
+        | TokenWatcherModeConfig
+    )
     local_mode: bool = False
     local_private_key: str | None = None
 
@@ -346,6 +417,12 @@ def __post_init__(self) -> None:
             self.mode_config, WatcherModeConfig
         ):
             raise ValueError("Watcher mode requires WatcherModeConfig")
+        elif self.oracle_mode == OracleMode.TOKEN_WATCHER and not isinstance(
+            self.mode_config, TokenWatcherModeConfig
+        ):
+            raise ValueError(
+                "Token watcher mode requires TokenWatcherModeConfig"
+            )
 
     @classmethod
     def from_env(cls, local_mode: bool = False) -> "OracleConfig":
@@ -395,15 +472,20 @@ def from_env(cls, local_mode: bool = False) -> "OracleConfig":
             oracle_mode = OracleMode.PUSH
         elif mode_str == "watcher":
             oracle_mode = OracleMode.WATCHER
+        elif mode_str == "token_watcher":
+            oracle_mode = OracleMode.TOKEN_WATCHER
         else:
             raise ValueError(
                 f"Invalid ORACLE_MODE: {mode_str}. "
-                "Must be one of: event_listener, push, watcher"
+                "Must be one of: event_listener, push, watcher, token_watcher"
             )
 
         # Load mode-specific configuration
         mode_config: (
-            EventListenerModeConfig | PushModeConfig | WatcherModeConfig
+            EventListenerModeConfig
+            | PushModeConfig
+            | WatcherModeConfig
+            | TokenWatcherModeConfig
         )
 
         if oracle_mode == OracleMode.EVENT_LISTENER:
@@ -417,7 +499,7 @@ def from_env(cls, local_mode: bool = False) -> "OracleConfig":
                 push_interval=int(os.environ.get("PUSH_INTERVAL", "60")),
                 batch_size=int(os.environ.get("PUSH_BATCH_SIZE", "20")),
             )
-        else:  # OracleMode.WATCHER
+        elif oracle_mode == OracleMode.WATCHER:
             watch_addresses_str = os.environ.get("WATCH_ADDRESSES", "")
             if not watch_addresses_str:
                 raise ValueError(
@@ -450,6 +532,46 @@ def from_env(cls, local_mode: bool = False) -> "OracleConfig":
                     os.environ.get("HEARTBEAT_INTERVAL_SECONDS", "3600")
                 ),
             )
+        else:  # OracleMode.TOKEN_WATCHER
+            token_addresses_str = os.environ.get("TOKEN_ADDRESSES", "")
+            if not token_addresses_str:
+                raise ValueError(
+                    "TOKEN_ADDRESSES is required for token watcher mode. "
+                    "Provide comma-separated token contract addresses (e.g., USDC, USDT)."
+                )
+
+            recipient_addresses_str = os.environ.get("RECIPIENT_ADDRESSES", "")
+            if not recipient_addresses_str:
+                raise ValueError(
+                    "RECIPIENT_ADDRESSES is required for token watcher mode. "
+                    "Provide comma-separated recipient addresses to watch."
+                )
+
+            # Parse comma-separated token addresses
+            token_addresses = [
+                addr.strip()
+                for addr in token_addresses_str.split(",")
+                if addr.strip()
+            ]
+
+            if len(token_addresses) == 0:
+                raise ValueError("TOKEN_ADDRESSES cannot be empty")
+
+            # Parse comma-separated recipient addresses
+            recipient_addresses = [
+                addr.strip()
+                for addr in recipient_addresses_str.split(",")
+                if addr.strip()
+            ]
+
+            if len(recipient_addresses) == 0:
+                raise ValueError("RECIPIENT_ADDRESSES cannot be empty")
+
+            mode_config = TokenWatcherModeConfig(
+                token_addresses=token_addresses,
+                recipient_addresses=recipient_addresses,
+                scan_interval=int(os.environ.get("SCAN_INTERVAL", "5")),
+            )
 
         # Load oracle-level config
         local_private_key = (
@@ -527,6 +649,26 @@ def log_config(self) -> None:
             for addr in self.mode_config.watch_addresses:
                 logger.info(f"    - {addr}")
 
+        elif self.oracle_mode == OracleMode.TOKEN_WATCHER:
+            assert isinstance(self.mode_config, TokenWatcherModeConfig)
+            logger.info("Token Watcher Settings:")
+            logger.info(
+                f"  Scan Interval: {self.mode_config.scan_interval} seconds"
+            )
+            logger.info(
+                "  Mode: Real-time (scans all new blocks since last check)"
+            )
+            logger.info(
+                f"  Watching {len(self.mode_config.token_addresses)} token(s):"
+            )
+            for addr in self.mode_config.token_addresses:
+                logger.info(f"    - {addr}")
+            logger.info(
+                f"  Monitoring {len(self.mode_config.recipient_addresses)} recipient(s):"
+            )
+            for addr in self.mode_config.recipient_addresses:
+                logger.info(f"    - {addr}")
+
         logger.info("Oracle Settings:")
         logger.info(f"  Mode: {'LOCAL' if self.local_mode else 'PRODUCTION'}")