feat: add foreign key constraints and metrics for blockchain node

apps/blockchain-node/src/aitbc_chain/app.py

@@ -1,24 +1,35 @@
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
-
 from fastapi import APIRouter, FastAPI
 from fastapi.responses import PlainTextResponse
 
 from .config import settings
 from .database import init_db
+from .gossip import create_backend, gossip_broker
 from .metrics import metrics_registry
 from .rpc.router import router as rpc_router
+from .rpc.websocket import router as websocket_router
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     init_db()
+    backend = create_backend(
+        settings.gossip_backend,
+        broadcast_url=settings.gossip_broadcast_url,
+    )
+    await gossip_broker.set_backend(backend)
+    try:
+        yield
+    finally:
+        await gossip_broker.shutdown()
+
 
 def create_app() -> FastAPI:
     app = FastAPI(title="AITBC Blockchain Node", version="0.1.0", lifespan=lifespan)
     app.include_router(rpc_router, prefix="/rpc", tags=["rpc"])
-
+    app.include_router(websocket_router, prefix="/rpc")
     metrics_router = APIRouter()
 
     @metrics_router.get("/metrics", response_class=PlainTextResponse, tags=["metrics"], summary="Prometheus metrics")

apps/blockchain-node/src/aitbc_chain/config.py

@@ -26,5 +26,8 @@ class ChainSettings(BaseSettings):
 
     block_time_seconds: int = 2
 
+    gossip_backend: str = "memory"
+    gossip_broadcast_url: Optional[str] = None
+
 
 settings = ChainSettings()

apps/blockchain-node/src/aitbc_chain/consensus/poa.py

@@ -11,6 +11,7 @@ from sqlmodel import Session, select
 from ..logging import get_logger
 from ..metrics import metrics_registry
 from ..models import Block
+from ..gossip import gossip_broker
 
 
 @dataclass
@@ -78,9 +79,11 @@ class PoAProposer:
             head = session.exec(select(Block).order_by(Block.height.desc()).limit(1)).first()
             next_height = 0
             parent_hash = "0x00"
+            interval_seconds: Optional[float] = None
             if head is not None:
                 next_height = head.height + 1
                 parent_hash = head.hash
+                interval_seconds = (datetime.utcnow() - head.timestamp).total_seconds()
 
             timestamp = datetime.utcnow()
             block_hash = self._compute_block_hash(next_height, parent_hash, timestamp)
@@ -99,6 +102,21 @@ class PoAProposer:
 
             metrics_registry.increment("blocks_proposed_total")
             metrics_registry.set_gauge("chain_head_height", float(next_height))
+            if interval_seconds is not None and interval_seconds >= 0:
+                metrics_registry.observe("block_interval_seconds", interval_seconds)
+
+            asyncio.create_task(
+                gossip_broker.publish(
+                    "blocks",
+                    {
+                        "height": block.height,
+                        "hash": block.hash,
+                        "parent_hash": block.parent_hash,
+                        "timestamp": block.timestamp.isoformat(),
+                        "tx_count": block.tx_count,
+                    },
+                )
+            )
 
             self._logger.info(
                 "Proposed block",
@@ -129,6 +147,19 @@ class PoAProposer:
             )
             session.add(genesis)
             session.commit()
+            asyncio.create_task(
+                gossip_broker.publish(
+                    "blocks",
+                    {
+                        "height": genesis.height,
+                        "hash": genesis.hash,
+                        "parent_hash": genesis.parent_hash,
+                        "timestamp": genesis.timestamp.isoformat(),
+                        "tx_count": genesis.tx_count,
+                    },
+                )
+            )
+
             self._logger.info("Created genesis block", extra={"hash": genesis_hash})
 
     def _fetch_chain_head(self) -> Optional[Block]:

apps/blockchain-node/src/aitbc_chain/gossip/__init__.py

@@ -0,0 +1,17 @@
+from .broker import (
+    BroadcastGossipBackend,
+    GossipBroker,
+    InMemoryGossipBackend,
+    TopicSubscription,
+    create_backend,
+    gossip_broker,
+)
+
+__all__ = [
+    "BroadcastGossipBackend",
+    "GossipBroker",
+    "InMemoryGossipBackend",
+    "TopicSubscription",
+    "create_backend",
+    "gossip_broker",
+]

apps/blockchain-node/src/aitbc_chain/gossip/broker.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from collections import defaultdict
+from contextlib import suppress
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, List, Optional, Set
+
+try:
+    from starlette.broadcast import Broadcast
+except ImportError:  # pragma: no cover - Starlette is an indirect dependency of FastAPI
+    Broadcast = None  # type: ignore[assignment]
+
+from ..metrics import metrics_registry
+
+
+def _increment_publication(metric_prefix: str, topic: str) -> None:
+    metrics_registry.increment(f"{metric_prefix}_total")
+    metrics_registry.increment(f"{metric_prefix}_topic_{topic}")
+
+
+def _set_queue_gauge(topic: str, size: int) -> None:
+    metrics_registry.set_gauge(f"gossip_queue_size_{topic}", float(size))
+
+
+def _update_subscriber_metrics(topics: Dict[str, List["asyncio.Queue[Any]"]]) -> None:
+    for topic, queues in topics.items():
+        metrics_registry.set_gauge(f"gossip_subscribers_topic_{topic}", float(len(queues)))
+    total = sum(len(queues) for queues in topics.values())
+    metrics_registry.set_gauge("gossip_subscribers_total", float(total))
+
+
+def _clear_topic_metrics(topic: str) -> None:
+    metrics_registry.set_gauge(f"gossip_subscribers_topic_{topic}", 0.0)
+    _set_queue_gauge(topic, 0)
+
+@dataclass
+class TopicSubscription:
+    topic: str
+    queue: "asyncio.Queue[Any]"
+    _unsubscribe: Callable[[], None]
+
+    def close(self) -> None:
+        self._unsubscribe()
+
+    async def get(self) -> Any:
+        return await self.queue.get()
+
+    async def __aiter__(self):  # type: ignore[override]
+        try:
+            while True:
+                yield await self.queue.get()
+        finally:
+            self.close()
+
+
+class GossipBackend:
+    async def start(self) -> None:  # pragma: no cover - overridden as needed
+        return None
+
+    async def publish(self, topic: str, message: Any) -> None:
+        raise NotImplementedError
+
+    async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
+        raise NotImplementedError
+
+    async def shutdown(self) -> None:
+        return None
+
+
+class InMemoryGossipBackend(GossipBackend):
+    def __init__(self) -> None:
+        self._topics: Dict[str, List["asyncio.Queue[Any]"]] = defaultdict(list)
+        self._lock = asyncio.Lock()
+
+    async def publish(self, topic: str, message: Any) -> None:
+        async with self._lock:
+            queues = list(self._topics.get(topic, []))
+        for queue in queues:
+            await queue.put(message)
+            _set_queue_gauge(topic, queue.qsize())
+        _increment_publication("gossip_publications", topic)
+
+    async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
+        queue: "asyncio.Queue[Any]" = asyncio.Queue(maxsize=max_queue_size)
+
+        async with self._lock:
+            self._topics[topic].append(queue)
+            _update_subscriber_metrics(self._topics)
+
+        _set_queue_gauge(topic, queue.qsize())
+
+        def _unsubscribe() -> None:
+            async def _remove() -> None:
+                async with self._lock:
+                    queues = self._topics.get(topic)
+                    if queues is None:
+                        return
+                    if queue in queues:
+                        queues.remove(queue)
+                        if not queues:
+                            self._topics.pop(topic, None)
+                            _clear_topic_metrics(topic)
+                        _update_subscriber_metrics(self._topics)
+
+            asyncio.create_task(_remove())
+
+        return TopicSubscription(topic=topic, queue=queue, _unsubscribe=_unsubscribe)
+
+    async def shutdown(self) -> None:
+        async with self._lock:
+            topics = list(self._topics.keys())
+            self._topics.clear()
+        for topic in topics:
+            _clear_topic_metrics(topic)
+        _update_subscriber_metrics(self._topics)
+
+
+class BroadcastGossipBackend(GossipBackend):
+    def __init__(self, url: str) -> None:
+        if Broadcast is None:  # pragma: no cover - dependency is optional
+            raise RuntimeError("Starlette Broadcast backend requested but starlette is not available")
+        self._broadcast = Broadcast(url)  # type: ignore[arg-type]
+        self._tasks: Set[asyncio.Task[None]] = set()
+        self._lock = asyncio.Lock()
+        self._running = False
+
+    async def start(self) -> None:
+        if not self._running:
+            await self._broadcast.connect()  # type: ignore[union-attr]
+            self._running = True
+
+    async def publish(self, topic: str, message: Any) -> None:
+        if not self._running:
+            raise RuntimeError("Broadcast backend not started")
+        payload = _encode_message(message)
+        await self._broadcast.publish(topic, payload)  # type: ignore[union-attr]
+        _increment_publication("gossip_broadcast_publications", topic)
+
+    async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
+        if not self._running:
+            raise RuntimeError("Broadcast backend not started")
+
+        queue: "asyncio.Queue[Any]" = asyncio.Queue(maxsize=max_queue_size)
+        stop_event = asyncio.Event()
+
+        async def _run_subscription() -> None:
+            async with self._broadcast.subscribe(topic) as subscriber:  # type: ignore[attr-defined,union-attr]
+                async for event in subscriber:  # type: ignore[union-attr]
+                    if stop_event.is_set():
+                        break
+                    data = _decode_message(getattr(event, "message", event))
+                    try:
+                        await queue.put(data)
+                        _set_queue_gauge(topic, queue.qsize())
+                    except asyncio.CancelledError:
+                        break
+
+        task = asyncio.create_task(_run_subscription(), name=f"broadcast-sub:{topic}")
+        async with self._lock:
+            self._tasks.add(task)
+            metrics_registry.set_gauge("gossip_broadcast_subscribers_total", float(len(self._tasks)))
+
+        def _unsubscribe() -> None:
+            async def _stop() -> None:
+                stop_event.set()
+                task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await task
+                async with self._lock:
+                    self._tasks.discard(task)
+                    metrics_registry.set_gauge("gossip_broadcast_subscribers_total", float(len(self._tasks)))
+
+            asyncio.create_task(_stop())
+
+        return TopicSubscription(topic=topic, queue=queue, _unsubscribe=_unsubscribe)
+
+    async def shutdown(self) -> None:
+        async with self._lock:
+            tasks = list(self._tasks)
+            self._tasks.clear()
+            metrics_registry.set_gauge("gossip_broadcast_subscribers_total", 0.0)
+        for task in tasks:
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task
+        if self._running:
+            await self._broadcast.disconnect()  # type: ignore[union-attr]
+            self._running = False
+
+
+class GossipBroker:
+    def __init__(self, backend: GossipBackend) -> None:
+        self._backend = backend
+        self._lock = asyncio.Lock()
+        self._started = False
+
+    async def publish(self, topic: str, message: Any) -> None:
+        if not self._started:
+            await self._backend.start()
+            self._started = True
+        await self._backend.publish(topic, message)
+
+    async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
+        if not self._started:
+            await self._backend.start()
+            self._started = True
+        return await self._backend.subscribe(topic, max_queue_size=max_queue_size)
+
+    async def set_backend(self, backend: GossipBackend) -> None:
+        await backend.start()
+        async with self._lock:
+            previous = self._backend
+            self._backend = backend
+            self._started = True
+        await previous.shutdown()
+
+    async def shutdown(self) -> None:
+        await self._backend.shutdown()
+        self._started = False
+        metrics_registry.set_gauge("gossip_subscribers_total", 0.0)
+
+
+def create_backend(backend_type: str, *, broadcast_url: Optional[str] = None) -> GossipBackend:
+    backend = backend_type.lower()
+    if backend in {"memory", "inmemory", "local"}:
+        return InMemoryGossipBackend()
+    if backend in {"broadcast", "starlette", "redis"}:
+        if not broadcast_url:
+            raise ValueError("Broadcast backend requires a gossip_broadcast_url setting")
+        return BroadcastGossipBackend(broadcast_url)
+    raise ValueError(f"Unsupported gossip backend '{backend_type}'")
+
+
+def _encode_message(message: Any) -> Any:
+    if isinstance(message, (str, bytes, bytearray)):
+        return message
+    return json.dumps(message, separators=(",", ":"))
+
+
+def _decode_message(message: Any) -> Any:
+    if isinstance(message, (bytes, bytearray)):
+        message = message.decode("utf-8")
+    if isinstance(message, str):
+        try:
+            return json.loads(message)
+        except json.JSONDecodeError:
+            return message
+    return message
+
+
+gossip_broker = GossipBroker(InMemoryGossipBackend())
+

apps/blockchain-node/src/aitbc_chain/metrics.py

@@ -15,6 +15,7 @@ class MetricsRegistry:
     def __init__(self) -> None:
         self._counters: Dict[str, float] = {}
         self._gauges: Dict[str, float] = {}
+        self._summaries: Dict[str, tuple[float, float]] = {}
         self._lock = Lock()
 
     def increment(self, name: str, amount: float = 1.0) -> None:
@@ -25,6 +26,17 @@ class MetricsRegistry:
         with self._lock:
             self._gauges[name] = value
 
+    def observe(self, name: str, value: float) -> None:
+        with self._lock:
+            count, total = self._summaries.get(name, (0.0, 0.0))
+            self._summaries[name] = (count + 1.0, total + value)
+
+    def reset(self) -> None:
+        with self._lock:
+            self._counters.clear()
+            self._gauges.clear()
+            self._summaries.clear()
+
     def render_prometheus(self) -> str:
         with self._lock:
             lines: list[str] = []
@@ -34,6 +46,10 @@ class MetricsRegistry:
             for name, value in sorted(self._gauges.items()):
                 lines.append(f"# TYPE {name} gauge")
                 lines.append(f"{name} {value}")
+            for name, (count, total) in sorted(self._summaries.items()):
+                lines.append(f"# TYPE {name} summary")
+                lines.append(f"{name}_count {count}")
+                lines.append(f"{name}_sum {total}")
             return "\n".join(lines) + "\n"
 
 

apps/blockchain-node/src/aitbc_chain/rpc/router.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import asyncio
 import json
+import time
 from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, HTTPException, status
@@ -8,6 +10,7 @@ from pydantic import BaseModel, Field, model_validator
 from sqlmodel import select
 
 from ..database import session_scope
+from ..gossip import gossip_broker
 from ..mempool import get_mempool
 from ..metrics import metrics_registry
 from ..models import Account, Block, Receipt, Transaction
@@ -64,84 +67,134 @@ class MintFaucetRequest(BaseModel):
 
 @router.get("/head", summary="Get current chain head")
 async def get_head() -> Dict[str, Any]:
+    metrics_registry.increment("rpc_get_head_total")
+    start = time.perf_counter()
     with session_scope() as session:
         result = session.exec(select(Block).order_by(Block.height.desc()).limit(1)).first()
         if result is None:
+            metrics_registry.increment("rpc_get_head_not_found_total")
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no blocks yet")
-        return {
-            "height": result.height,
-            "hash": result.hash,
-            "timestamp": result.timestamp.isoformat(),
-            "tx_count": result.tx_count,
-        }
+        metrics_registry.increment("rpc_get_head_success_total")
+    metrics_registry.observe("rpc_get_head_duration_seconds", time.perf_counter() - start)
+    return {
+        "height": result.height,
+        "hash": result.hash,
+        "timestamp": result.timestamp.isoformat(),
+        "tx_count": result.tx_count,
+    }
 
 
 @router.get("/blocks/{height}", summary="Get block by height")
 async def get_block(height: int) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_get_block_total")
+    start = time.perf_counter()
     with session_scope() as session:
         block = session.exec(select(Block).where(Block.height == height)).first()
         if block is None:
+            metrics_registry.increment("rpc_get_block_not_found_total")
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="block not found")
-        return {
-            "height": block.height,
-            "hash": block.hash,
-            "parent_hash": block.parent_hash,
-            "timestamp": block.timestamp.isoformat(),
-            "tx_count": block.tx_count,
-            "state_root": block.state_root,
-        }
+        metrics_registry.increment("rpc_get_block_success_total")
+    metrics_registry.observe("rpc_get_block_duration_seconds", time.perf_counter() - start)
+    return {
+        "height": block.height,
+        "hash": block.hash,
+        "parent_hash": block.parent_hash,
+        "timestamp": block.timestamp.isoformat(),
+        "tx_count": block.tx_count,
+        "state_root": block.state_root,
+    }
 
 
 @router.get("/tx/{tx_hash}", summary="Get transaction by hash")
 async def get_transaction(tx_hash: str) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_get_transaction_total")
+    start = time.perf_counter()
     with session_scope() as session:
         tx = session.exec(select(Transaction).where(Transaction.tx_hash == tx_hash)).first()
         if tx is None:
+            metrics_registry.increment("rpc_get_transaction_not_found_total")
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="transaction not found")
-        return {
-            "tx_hash": tx.tx_hash,
-            "block_height": tx.block_height,
-            "sender": tx.sender,
-            "recipient": tx.recipient,
-            "payload": tx.payload,
-            "created_at": tx.created_at.isoformat(),
-        }
+        metrics_registry.increment("rpc_get_transaction_success_total")
+    metrics_registry.observe("rpc_get_transaction_duration_seconds", time.perf_counter() - start)
+    return {
+        "tx_hash": tx.tx_hash,
+        "block_height": tx.block_height,
+        "sender": tx.sender,
+        "recipient": tx.recipient,
+        "payload": tx.payload,
+        "created_at": tx.created_at.isoformat(),
+    }
 
 
 @router.get("/receipts/{receipt_id}", summary="Get receipt by ID")
 async def get_receipt(receipt_id: str) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_get_receipt_total")
+    start = time.perf_counter()
     with session_scope() as session:
         receipt = session.exec(select(Receipt).where(Receipt.receipt_id == receipt_id)).first()
         if receipt is None:
+            metrics_registry.increment("rpc_get_receipt_not_found_total")
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="receipt not found")
-        return _serialize_receipt(receipt)
+        metrics_registry.increment("rpc_get_receipt_success_total")
+    metrics_registry.observe("rpc_get_receipt_duration_seconds", time.perf_counter() - start)
+    return _serialize_receipt(receipt)
 
 
 @router.get("/getBalance/{address}", summary="Get account balance")
 async def get_balance(address: str) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_get_balance_total")
+    start = time.perf_counter()
     with session_scope() as session:
         account = session.get(Account, address)
         if account is None:
+            metrics_registry.increment("rpc_get_balance_empty_total")
+            metrics_registry.observe("rpc_get_balance_duration_seconds", time.perf_counter() - start)
             return {"address": address, "balance": 0, "nonce": 0}
-        return {
-            "address": account.address,
-            "balance": account.balance,
-            "nonce": account.nonce,
-            "updated_at": account.updated_at.isoformat(),
-        }
+        metrics_registry.increment("rpc_get_balance_success_total")
+    metrics_registry.observe("rpc_get_balance_duration_seconds", time.perf_counter() - start)
+    return {
+        "address": account.address,
+        "balance": account.balance,
+        "nonce": account.nonce,
+        "updated_at": account.updated_at.isoformat(),
+    }
 
 
 @router.post("/sendTx", summary="Submit a new transaction")
 async def send_transaction(request: TransactionRequest) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_send_tx_total")
+    start = time.perf_counter()
     mempool = get_mempool()
     tx_dict = request.model_dump()
     tx_hash = mempool.add(tx_dict)
-    metrics_registry.increment("rpc_send_tx_total")
-    return {"tx_hash": tx_hash}
+    try:
+        asyncio.create_task(
+            gossip_broker.publish(
+                "transactions",
+                {
+                    "tx_hash": tx_hash,
+                    "sender": request.sender,
+                    "recipient": request.recipient,
+                    "payload": request.payload,
+                    "nonce": request.nonce,
+                    "fee": request.fee,
+                    "type": request.type,
+                },
+            )
+        )
+        metrics_registry.increment("rpc_send_tx_success_total")
+        return {"tx_hash": tx_hash}
+    except Exception:
+        metrics_registry.increment("rpc_send_tx_failed_total")
+        raise
+    finally:
+        metrics_registry.observe("rpc_send_tx_duration_seconds", time.perf_counter() - start)
 
 
 @router.post("/submitReceipt", summary="Submit receipt claim transaction")
 async def submit_receipt(request: ReceiptSubmissionRequest) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_submit_receipt_total")
+    start = time.perf_counter()
     tx_payload = {
         "type": "RECEIPT_CLAIM",
         "sender": request.sender,
@@ -151,17 +204,31 @@ async def submit_receipt(request: ReceiptSubmissionRequest) -> Dict[str, Any]:
         "sig": request.sig,
     }
     tx_request = TransactionRequest.model_validate(tx_payload)
-    metrics_registry.increment("rpc_submit_receipt_total")
-    return await send_transaction(tx_request)
+    try:
+        response = await send_transaction(tx_request)
+        metrics_registry.increment("rpc_submit_receipt_success_total")
+        return response
+    except HTTPException:
+        metrics_registry.increment("rpc_submit_receipt_failed_total")
+        raise
+    except Exception:
+        metrics_registry.increment("rpc_submit_receipt_failed_total")
+        raise
+    finally:
+        metrics_registry.observe("rpc_submit_receipt_duration_seconds", time.perf_counter() - start)
 
 
 @router.post("/estimateFee", summary="Estimate transaction fee")
 async def estimate_fee(request: EstimateFeeRequest) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_estimate_fee_total")
+    start = time.perf_counter()
     base_fee = 10
     per_byte = 1
     payload_bytes = len(json.dumps(request.payload, sort_keys=True, separators=(",", ":")).encode())
     estimated_fee = base_fee + per_byte * payload_bytes
     tx_type = (request.type or "TRANSFER").upper()
+    metrics_registry.increment("rpc_estimate_fee_success_total")
+    metrics_registry.observe("rpc_estimate_fee_duration_seconds", time.perf_counter() - start)
     return {
         "type": tx_type,
         "base_fee": base_fee,
@@ -172,6 +239,8 @@ async def estimate_fee(request: EstimateFeeRequest) -> Dict[str, Any]:
 
 @router.post("/admin/mintFaucet", summary="Mint devnet funds to an address")
 async def mint_faucet(request: MintFaucetRequest) -> Dict[str, Any]:
+    metrics_registry.increment("rpc_mint_faucet_total")
+    start = time.perf_counter()
     with session_scope() as session:
         account = session.get(Account, request.address)
         if account is None:
@@ -181,4 +250,6 @@ async def mint_faucet(request: MintFaucetRequest) -> Dict[str, Any]:
             account.balance += request.amount
         session.commit()
         updated_balance = account.balance
+    metrics_registry.increment("rpc_mint_faucet_success_total")
+    metrics_registry.observe("rpc_mint_faucet_duration_seconds", time.perf_counter() - start)
     return {"address": request.address, "balance": updated_balance}

apps/blockchain-node/src/aitbc_chain/rpc/websocket.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import asyncio
+from typing import AsyncIterator, Dict
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from ..gossip import gossip_broker
+
+router = APIRouter(prefix="/ws", tags=["ws"])
+
+
+async def _stream_topic(topic: str, websocket: WebSocket) -> None:
+    subscription = await gossip_broker.subscribe(topic)
+    try:
+        while True:
+            message = await subscription.get()
+            await websocket.send_json(message)
+    except WebSocketDisconnect:
+        pass
+    finally:
+        subscription.close()
+
+
+@router.websocket("/blocks")
+async def blocks_stream(websocket: WebSocket) -> None:
+    await websocket.accept()
+    await _stream_topic("blocks", websocket)
+
+
+@router.websocket("/transactions")
+async def transactions_stream(websocket: WebSocket) -> None:
+    await websocket.accept()
+    await _stream_topic("transactions", websocket)