Add blockchain event bridge service with smart contract event integration

- Phase 1: Core bridge service with gossip broker subscription
- Phase 2: Smart contract event integration via eth_getLogs RPC endpoint
- Add contract event subscriber for AgentStaking, PerformanceVerifier, Marketplace, Bounty, CrossChainBridge
- Add contract event handlers in agent_daemon.py and marketplace.py
- Add systemd service file for blockchain-event-bridge
- Update blockchain node router.py with eth_getLogs endpoint
- Add configuration for contract addresses
- Add tests for contract subscriber and handlers (27 tests passing)

apps/blockchain-event-bridge/src/blockchain_event_bridge/event_subscribers/__init__.py

@@ -0,0 +1,7 @@
+"""Event subscriber modules for blockchain events."""
+
+from .blocks import BlockEventSubscriber
+from .transactions import TransactionEventSubscriber
+from .contracts import ContractEventSubscriber
+
+__all__ = ["BlockEventSubscriber", "TransactionEventSubscriber", "ContractEventSubscriber"]

apps/blockchain-event-bridge/src/blockchain_event_bridge/event_subscribers/blocks.py

@@ -0,0 +1,107 @@
+"""Block event subscriber for gossip broker."""
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any, Dict
+
+from ..config import Settings
+from ..metrics import event_queue_size, gossip_subscribers_total
+
+if TYPE_CHECKING:
+    from ..bridge import BlockchainEventBridge
+
+
+logger = logging.getLogger(__name__)
+
+
+class BlockEventSubscriber:
+    """Subscribes to block events from the gossip broker."""
+
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+        self._running = False
+        self._bridge: "BlockchainEventBridge | None" = None
+        self._subscription = None
+
+    def set_bridge(self, bridge: "BlockchainEventBridge") -> None:
+        """Set the bridge instance for event handling."""
+        self._bridge = bridge
+
+    async def run(self) -> None:
+        """Run the block event subscriber."""
+        if self._running:
+            logger.warning("Block event subscriber already running")
+            return
+
+        self._running = True
+        logger.info("Starting block event subscriber...")
+
+        # Import gossip broker from blockchain-node
+        try:
+            # Add blockchain-node to path for import
+            import sys
+            from pathlib import Path
+            blockchain_node_src = Path("/opt/aitbc/apps/blockchain-node/src")
+            if str(blockchain_node_src) not in sys.path:
+                sys.path.insert(0, str(blockchain_node_src))
+
+            from aitbc_chain.gossip.broker import create_backend, GossipBroker
+
+            # Create gossip backend
+            backend = create_backend(
+                self.settings.gossip_backend,
+                broadcast_url=self.settings.gossip_broadcast_url
+            )
+            self._broker = GossipBroker(backend)
+
+            # Subscribe to blocks topic
+            self._subscription = await self._broker.subscribe("blocks", max_queue_size=100)
+            gossip_subscribers_total.set(1)
+
+            logger.info("Successfully subscribed to blocks topic")
+
+        except ImportError as e:
+            logger.error(f"Failed to import gossip broker: {e}")
+            logger.info("Using mock implementation for development")
+            await self._run_mock()
+            return
+
+        # Process block events
+        while self._running:
+            try:
+                block_data = await self._subscription.get()
+                event_queue_size.labels(topic="blocks").set(self._subscription.queue.qsize())
+
+                logger.info(f"Received block event: height={block_data.get('height')}")
+
+                if self._bridge:
+                    await self._bridge.handle_block_event(block_data)
+
+            except asyncio.CancelledError:
+                logger.info("Block event subscriber cancelled")
+                break
+            except Exception as e:
+                logger.error(f"Error processing block event: {e}", exc_info=True)
+                await asyncio.sleep(1)  # Avoid tight error loop
+
+    async def _run_mock(self) -> None:
+        """Run a mock subscriber for development/testing when gossip broker is unavailable."""
+        logger.warning("Using mock block event subscriber - no real events will be processed")
+        await asyncio.sleep(60)  # Keep alive but do nothing
+
+    async def stop(self) -> None:
+        """Stop the block event subscriber."""
+        if not self._running:
+            return
+
+        logger.info("Stopping block event subscriber...")
+        self._running = False
+
+        if self._subscription:
+            self._subscription.close()
+
+        if hasattr(self, '_broker'):
+            await self._broker.shutdown()
+
+        gossip_subscribers_total.set(0)
+        logger.info("Block event subscriber stopped")

apps/blockchain-event-bridge/src/blockchain_event_bridge/event_subscribers/contracts.py

@@ -0,0 +1,223 @@
+"""Contract event subscriber for smart contract event monitoring."""
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+import httpx
+
+from ..config import Settings
+from ..metrics import event_queue_size
+
+if TYPE_CHECKING:
+    from ..bridge import BlockchainEventBridge
+
+
+logger = logging.getLogger(__name__)
+
+
+class ContractEventSubscriber:
+    """Subscribes to smart contract events via blockchain RPC."""
+
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+        self._running = False
+        self._bridge: "BlockchainEventBridge | None" = None
+        self._client: Optional[httpx.AsyncClient] = None
+
+        # Contract addresses from configuration
+        self.contract_addresses: Dict[str, str] = {
+            "AgentStaking": settings.agent_staking_address or "",
+            "PerformanceVerifier": settings.performance_verifier_address or "",
+            "AgentServiceMarketplace": settings.marketplace_address or "",
+            "BountyIntegration": settings.bounty_address or "",
+            "CrossChainBridge": settings.bridge_address or "",
+        }
+
+        # Event topics/signatures for each contract
+        self.event_topics: Dict[str, list[str]] = {
+            "AgentStaking": [
+                "StakeCreated",
+                "RewardsDistributed",
+                "AgentTierUpdated",
+            ],
+            "PerformanceVerifier": [
+                "PerformanceVerified",
+                "PenaltyApplied",
+                "RewardIssued",
+            ],
+            "AgentServiceMarketplace": [
+                "ServiceListed",
+                "ServicePurchased",
+            ],
+            "BountyIntegration": [
+                "BountyCreated",
+                "BountyCompleted",
+            ],
+            "CrossChainBridge": [
+                "BridgeInitiated",
+                "BridgeCompleted",
+            ],
+        }
+
+        # Track last processed block for each contract
+        self.last_processed_blocks: Dict[str, int] = {}
+
+    def set_bridge(self, bridge: "BlockchainEventBridge") -> None:
+        """Set the bridge instance for event handling."""
+        self._bridge = bridge
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        """Get or create HTTP client."""
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                base_url=self.settings.blockchain_rpc_url,
+                timeout=30.0,
+            )
+        return self._client
+
+    async def run(self) -> None:
+        """Run the contract event subscriber."""
+        if not self.settings.subscribe_contracts:
+            logger.info("Contract event subscription disabled")
+            return
+
+        if self._running:
+            logger.warning("Contract event subscriber already running")
+            return
+
+        self._running = True
+        logger.info("Starting contract event subscriber...")
+
+        # Initialize last processed blocks from current chain height
+        await self._initialize_block_tracking()
+
+        while self._running:
+            try:
+                await self._poll_contract_events()
+                await asyncio.sleep(self.settings.polling_interval_seconds)
+            except asyncio.CancelledError:
+                logger.info("Contract event subscriber cancelled")
+                break
+            except Exception as e:
+                logger.error(f"Error in contract event subscriber: {e}", exc_info=True)
+                await asyncio.sleep(5)
+
+    async def _initialize_block_tracking(self) -> None:
+        """Initialize block tracking from current chain height."""
+        try:
+            client = await self._get_client()
+            response = await client.get("/head")
+            if response.status_code == 200:
+                head_data = response.json()
+                current_height = head_data.get("height", 0)
+                for contract in self.contract_addresses:
+                    if self.contract_addresses[contract]:
+                        self.last_processed_blocks[contract] = current_height
+                logger.info(f"Initialized block tracking at height {current_height}")
+        except Exception as e:
+            logger.error(f"Error initializing block tracking: {e}")
+
+    async def _poll_contract_events(self) -> None:
+        """Poll for contract events from blockchain."""
+        client = await self._get_client()
+
+        for contract_name, contract_address in self.contract_addresses.items():
+            if not contract_address:
+                continue
+
+            try:
+                # Get current chain height
+                response = await client.get("/head")
+                if response.status_code != 200:
+                    logger.error(f"Failed to get chain head: {response.status_code}")
+                    continue
+
+                head_data = response.json()
+                current_height = head_data.get("height", 0)
+                last_height = self.last_processed_blocks.get(contract_name, current_height - 100)
+
+                # Query events for this contract
+                logs_response = await client.post(
+                    "/eth_getLogs",
+                    json={
+                        "address": contract_address,
+                        "from_block": last_height + 1,
+                        "to_block": current_height,
+                        "topics": self.event_topics.get(contract_name, []),
+                    }
+                )
+
+                if logs_response.status_code != 200:
+                    logger.error(f"Failed to get logs for {contract_name}: {logs_response.status_code}")
+                    continue
+
+                logs_data = logs_response.json()
+                logs = logs_data.get("logs", [])
+
+                if logs:
+                    logger.info(f"Found {len(logs)} events for {contract_name}")
+
+                    # Process each log
+                    for log in logs:
+                        await self._process_contract_event(contract_name, log)
+
+                # Update last processed block
+                self.last_processed_blocks[contract_name] = current_height
+
+            except Exception as e:
+                logger.error(f"Error polling events for {contract_name}: {e}", exc_info=True)
+
+    async def _process_contract_event(self, contract_name: str, log: Dict[str, Any]) -> None:
+        """Process a contract event."""
+        event_type = log.get("topics", [""])[0] if log.get("topics") else "Unknown"
+
+        logger.info(f"Processing {contract_name} event: {event_type}")
+
+        if self._bridge:
+            # Route event to appropriate handler based on contract type
+            if contract_name == "AgentStaking":
+                await self._handle_staking_event(log)
+            elif contract_name == "PerformanceVerifier":
+                await self._handle_performance_event(log)
+            elif contract_name == "AgentServiceMarketplace":
+                await self._handle_marketplace_event(log)
+            elif contract_name == "BountyIntegration":
+                await self._handle_bounty_event(log)
+            elif contract_name == "CrossChainBridge":
+                await self._handle_bridge_event(log)
+
+    async def _handle_staking_event(self, log: Dict[str, Any]) -> None:
+        """Handle AgentStaking contract event."""
+        if self._bridge:
+            await self._bridge.handle_staking_event(log)
+
+    async def _handle_performance_event(self, log: Dict[str, Any]) -> None:
+        """Handle PerformanceVerifier contract event."""
+        if self._bridge:
+            await self._bridge.handle_performance_event(log)
+
+    async def _handle_marketplace_event(self, log: Dict[str, Any]) -> None:
+        """Handle AgentServiceMarketplace contract event."""
+        if self._bridge:
+            await self._bridge.handle_marketplace_event(log)
+
+    async def _handle_bounty_event(self, log: Dict[str, Any]) -> None:
+        """Handle BountyIntegration contract event."""
+        if self._bridge:
+            await self._bridge.handle_bounty_event(log)
+
+    async def _handle_bridge_event(self, log: Dict[str, Any]) -> None:
+        """Handle CrossChainBridge contract event."""
+        if self._bridge:
+            await self._bridge.handle_bridge_event(log)
+
+    async def stop(self) -> None:
+        """Stop the contract event subscriber."""
+        self._running = False
+
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+
+        logger.info("Contract event subscriber stopped")

apps/blockchain-event-bridge/src/blockchain_event_bridge/event_subscribers/transactions.py

@@ -0,0 +1,113 @@
+"""Transaction event subscriber for gossip broker."""
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any, Dict
+
+from ..config import Settings
+from ..metrics import event_queue_size, gossip_subscribers_total
+
+if TYPE_CHECKING:
+    from ..bridge import BlockchainEventBridge
+
+
+logger = logging.getLogger(__name__)
+
+
+class TransactionEventSubscriber:
+    """Subscribes to transaction events from the gossip broker."""
+
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+        self._running = False
+        self._bridge: "BlockchainEventBridge | None" = None
+        self._subscription = None
+
+    def set_bridge(self, bridge: "BlockchainEventBridge") -> None:
+        """Set the bridge instance for event handling."""
+        self._bridge = bridge
+
+    async def run(self) -> None:
+        """Run the transaction event subscriber."""
+        if self._running:
+            logger.warning("Transaction event subscriber already running")
+            return
+
+        self._running = True
+        logger.info("Starting transaction event subscriber...")
+
+        # Import gossip broker from blockchain-node
+        try:
+            # Add blockchain-node to path for import
+            import sys
+            from pathlib import Path
+            blockchain_node_src = Path("/opt/aitbc/apps/blockchain-node/src")
+            if str(blockchain_node_src) not in sys.path:
+                sys.path.insert(0, str(blockchain_node_src))
+
+            from aitbc_chain.gossip.broker import create_backend, GossipBroker
+
+            # Create gossip backend
+            backend = create_backend(
+                self.settings.gossip_backend,
+                broadcast_url=self.settings.gossip_broadcast_url
+            )
+            self._broker = GossipBroker(backend)
+
+            # Subscribe to transactions topic (if available)
+            # Note: Currently transactions are embedded in block events
+            # This subscriber will be enhanced when transaction events are published separately
+            try:
+                self._subscription = await self._broker.subscribe("transactions", max_queue_size=100)
+                gossip_subscribers_total.inc()
+                logger.info("Successfully subscribed to transactions topic")
+            except Exception:
+                logger.info("Transactions topic not available - will extract from block events")
+                await self._run_mock()
+                return
+
+        except ImportError as e:
+            logger.error(f"Failed to import gossip broker: {e}")
+            logger.info("Using mock implementation for development")
+            await self._run_mock()
+            return
+
+        # Process transaction events
+        while self._running:
+            try:
+                tx_data = await self._subscription.get()
+                event_queue_size.labels(topic="transactions").set(self._subscription.queue.qsize())
+
+                logger.info(f"Received transaction event: hash={tx_data.get('hash')}")
+
+                if self._bridge:
+                    await self._bridge.handle_transaction_event(tx_data)
+
+            except asyncio.CancelledError:
+                logger.info("Transaction event subscriber cancelled")
+                break
+            except Exception as e:
+                logger.error(f"Error processing transaction event: {e}", exc_info=True)
+                await asyncio.sleep(1)
+
+    async def _run_mock(self) -> None:
+        """Run a mock subscriber for development/testing."""
+        logger.warning("Using mock transaction event subscriber - no real events will be processed")
+        await asyncio.sleep(60)
+
+    async def stop(self) -> None:
+        """Stop the transaction event subscriber."""
+        if not self._running:
+            return
+
+        logger.info("Stopping transaction event subscriber...")
+        self._running = False
+
+        if self._subscription:
+            self._subscription.close()
+
+        if hasattr(self, '_broker'):
+            await self._broker.shutdown()
+
+        gossip_subscribers_total.dec()
+        logger.info("Transaction event subscriber stopped")