Basic block explorer with faked data.

.gitignore

@@ -0,0 +1,5 @@
+.venv/**
+.idea/**
+**/__pycache__/**
+sqlite.db
+*.ignore*

.pre-commit-config.yaml

@@ -0,0 +1,18 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v6.0.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+
+  - repo: https://github.com/psf/black
+    rev: 25.9.0
+    hooks:
+      - id: black
+
+  - repo: https://github.com/PyCQA/isort
+    rev: 6.1.0
+    hooks:
+      - id: isort

README.md

@@ -0,0 +1 @@
+# Nomos Block Explorer

pyproject.toml

@@ -0,0 +1,11 @@
+[tool.black]
+line-length = 120
+target-version = ["py313"]
+skip-string-normalization = false
+
+[tool.isort]
+profile = "black"
+combine_as_imports = true
+src_paths = ["src"]
+skip_gitignore = true
+ensure_newline_before_comments = true

requirements.txt

@@ -0,0 +1,6 @@
+fastapi~=0.118.0
+uvicorn~=0.37.0
+requests~=2.32.5
+python-on-whales~=0.78.0
+sqlmodel~=0.0.25
+rusty-results~=1.1.1

src/__init__.py

@@ -0,0 +1,4 @@
+from pathlib import Path
+
+DIR_SRC = Path(__file__).resolve().parent
+DIR_REPO = DIR_SRC.parent

src/api/router.py

@@ -0,0 +1,9 @@
+from fastapi import APIRouter
+
+from .v1.router import router as v1_router
+
+
+def create_api_router() -> APIRouter:
+    router = APIRouter()
+    router.include_router(v1_router, prefix="/v1")
+    return router

src/api/utils.py

@@ -0,0 +1,27 @@
+from asyncio import sleep
+from typing import AsyncIterable, AsyncIterator, List, Union
+
+from core.models import IdNbeModel
+
+Data = Union[IdNbeModel, List[IdNbeModel]]
+Stream = AsyncIterator[Data]
+
+
+def _parse_stream_data(data: Data) -> bytes:
+    if isinstance(data, list):
+        return b"".join(item.model_dump_ndjson() for item in data)
+    else:
+        return data.model_dump_ndjson()
+
+
+async def streamer(
+    stream: Stream,
+    bootstrap_data: Data = None,
+    interval: int = 5,
+) -> AsyncIterable[bytes]:
+    if bootstrap_data is not None:
+        yield _parse_stream_data(bootstrap_data)
+
+    async for data in stream:
+        yield _parse_stream_data(data)
+        await sleep(interval)

src/api/v1/blocks.py

@@ -0,0 +1,15 @@
+from typing import List
+
+from starlette.responses import Response
+
+from api.utils import streamer
+from core.api import NBERequest, NDJsonStreamingResponse
+from node.models.blocks import Block
+
+
+async def stream(request: NBERequest) -> Response:
+    bootstrap_blocks: List[Block] = await request.app.state.block_repository.get_latest(limit=5, descending=False)
+    highest_slot: int = max((block.slot for block in bootstrap_blocks), default=0)
+    updates_stream = request.app.state.block_repository.updates_stream(slot_from=highest_slot + 1)
+    block_stream = streamer(stream=updates_stream, bootstrap_data=bootstrap_blocks)
+    return NDJsonStreamingResponse(block_stream)

src/api/v1/health.py

@@ -0,0 +1,24 @@
+from typing import AsyncIterator
+
+from starlette.responses import JSONResponse, Response
+
+from api.utils import streamer
+from core.api import NBERequest, NDJsonStreamingResponse
+from node.api.base import NodeApi
+from node.models.health import Health
+
+
+async def get(request: NBERequest) -> Response:
+    response = await request.app.state.node_api.get_health_check()
+    return JSONResponse(response)
+
+
+async def _health_iterator(node_api: NodeApi) -> AsyncIterator[Health]:
+    while True:
+        yield await node_api.get_health_check()
+
+
+async def stream(request: NBERequest) -> Response:
+    _stream = _health_iterator(request.app.state.node_api)
+    health_stream = streamer(stream=_stream, interval=2)
+    return NDJsonStreamingResponse(health_stream)

src/api/v1/index.py

@@ -0,0 +1,8 @@
+from starlette.responses import JSONResponse, Response
+
+from core.api import NBERequest
+
+
+async def index(_request: NBERequest) -> Response:
+    content = {"version": "1"}
+    return JSONResponse(content)

src/api/v1/router.py

@@ -0,0 +1,11 @@
+from fastapi import APIRouter
+
+from . import blocks, health, index, transactions
+
+router = APIRouter()
+router.add_api_route("/", index.index, methods=["GET", "HEAD"])
+router.add_api_route("/health", health.get, methods=["GET", "HEAD"])
+router.add_api_route("/health/stream", health.stream, methods=["GET", "HEAD"])
+
+router.add_api_route("/transactions/stream", transactions.stream, methods=["GET"])
+router.add_api_route("/blocks/stream", blocks.stream, methods=["GET"])

src/api/v1/transactions.py

@@ -0,0 +1,23 @@
+from datetime import datetime
+from typing import List
+
+from starlette.responses import Response
+
+from api.utils import streamer
+from core.api import NBERequest, NDJsonStreamingResponse
+from node.models.transactions import Transaction
+from utils.datetime import increment_datetime
+
+
+async def stream(request: NBERequest) -> Response:
+    bootstrap_transactions: List[Transaction] = await request.app.state.transaction_repository.get_latest(
+        limit=5, descending=False
+    )
+    highest_timestamp: datetime = max(
+        (transaction.timestamp for transaction in bootstrap_transactions), default=datetime.min
+    )
+    updates_stream = request.app.state.transaction_repository.updates_stream(
+        timestamp_from=increment_datetime(highest_timestamp)
+    )
+    transaction_stream = streamer(stream=updates_stream, bootstrap_data=bootstrap_transactions)
+    return NDJsonStreamingResponse(transaction_stream)

src/app.py

@@ -0,0 +1,11 @@
+from fastapi import FastAPI
+
+from core.app import NBE
+from lifespan import lifespan
+from router import create_router
+
+
+def create_app() -> FastAPI:
+    app = NBE(lifespan=lifespan)
+    app.include_router(create_router())
+    return app

src/core/api.py

@@ -0,0 +1,21 @@
+from starlette.requests import Request
+from starlette.responses import ContentStream, StreamingResponse
+
+from core.app import NBE
+
+
+class NBERequest(Request):
+    app: NBE
+
+
+class NDJsonStreamingResponse(StreamingResponse):
+    def __init__(self, content: ContentStream):
+        super().__init__(
+            content,
+            media_type="application/x-ndjson",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )

src/core/app.py

@@ -0,0 +1,45 @@
+from asyncio import Task, gather
+from typing import Optional
+
+from fastapi import FastAPI
+from starlette.datastructures import State
+
+from db.blocks import BlockRepository
+from db.clients import DbClient
+from db.transaction import TransactionRepository
+from node.api.base import NodeApi
+from node.manager.base import NodeManager
+
+
+class NBEState(State):
+    signal_exit: bool = False
+    node_manager: Optional[NodeManager]
+    node_api: Optional[NodeApi]
+    db_client: DbClient
+    block_repository: BlockRepository
+    transaction_repository: TransactionRepository
+    subscription_to_updates_handle: Task
+    backfill_handle: Task
+
+    @property
+    def is_running(self) -> bool:
+        return not self.signal_exit
+
+    async def stop(self):
+        self.signal_exit = True
+        await self._wait_tasks_finished()
+
+    async def _wait_tasks_finished(self):
+        await gather(
+            self.subscription_to_updates_handle,
+            self.backfill_handle,
+            return_exceptions=True,
+        )
+
+
+class NBE(FastAPI):
+    state: NBEState
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.state = NBEState()

src/core/encodings.py

@@ -0,0 +1,5 @@
+import json
+
+
+def ndjson(json_data: dict | list) -> bytes:
+    return (json.dumps(json_data) + "\n").encode("utf-8")

src/core/models.py

@@ -0,0 +1,12 @@
+from typing import Optional
+
+from sqlmodel import Field, SQLModel
+
+
+class NbeModel(SQLModel):
+    def model_dump_ndjson(self) -> bytes:
+        return f"{self.model_dump_json()}\n".encode("utf-8")
+
+
+class IdNbeModel(NbeModel):
+    id: Optional[int] = Field(default=None, primary_key=True)

src/db/blocks.py

@@ -0,0 +1,52 @@
+from typing import AsyncIterator, Iterable, List
+
+from rusty_results import Empty, Option, Some
+from sqlalchemy import Result
+from sqlmodel import select
+
+from db.clients import DbClient
+from node.models.blocks import Block
+
+
+class BlockRepository:
+    def __init__(self, client: DbClient):
+        self.client = client
+
+    async def create(self, block: Iterable[Block]) -> None:
+        with self.client.session() as session:
+            session.add_all(block)
+            session.commit()
+
+    async def get_latest(self, limit: int, descending: bool = True) -> List[Block]:
+        statement = select(Block).limit(limit)
+        if descending:
+            statement = statement.order_by(Block.slot.desc())
+        else:
+            statement = statement.order_by(Block.slot.asc())
+
+        with self.client.session() as session:
+            results: Result[Block] = session.exec(statement)
+            return results.all()
+
+    async def updates_stream(self, slot_from: int) -> AsyncIterator[List[Block]]:
+        _slot_from = slot_from
+        while True:
+            statement = select(Block).where(Block.slot >= _slot_from).order_by(Block.slot.asc())
+
+            with self.client.session() as session:
+                blocks: List[Block] = session.exec(statement).all()
+
+            if len(blocks) > 0:
+                # POC: Assumes slots are sequential and one block per slot
+                _slot_from = blocks[-1].slot + 1
+
+            yield blocks
+
+    async def get_earliest(self) -> Option[Block]:
+        with self.client.session() as session:
+            statement = select(Block).order_by(Block.slot.asc()).limit(1)
+            results: Result[Block] = session.exec(statement)
+            if (block := results.first()) is not None:
+                return Some(block)
+            else:
+                return Empty()

src/db/clients/__init__.py

@@ -0,0 +1,2 @@
+from .base import DbClient
+from .sqlite import SqliteClient

src/db/clients/base.py

@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+from typing import Iterator
+
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+class DbClient(ABC):
+    @abstractmethod
+    def connect(self):
+        pass
+
+    @abstractmethod
+    def session(self) -> Iterator[AsyncSession]:
+        pass
+
+    @abstractmethod
+    def disconnect(self):
+        pass

src/db/clients/sqlite.py

@@ -0,0 +1,28 @@
+from contextlib import contextmanager
+from typing import Iterator
+
+from sqlalchemy.engine.base import Engine
+from sqlmodel import Session, SQLModel, create_engine
+
+from db.clients.base import DbClient
+from src import DIR_REPO
+
+SQLITE_DB_PATH = DIR_REPO.joinpath("sqlite.db")
+
+
+# TODO: Async
+class SqliteClient(DbClient):
+    def __init__(self, sqlite_db_path: str = f"sqlite:///{SQLITE_DB_PATH}") -> None:
+        self.engine: Engine = create_engine(sqlite_db_path)
+        SQLModel.metadata.create_all(self.engine)
+
+    def connect(self):
+        pass
+
+    @contextmanager
+    def session(self) -> Iterator[Session]:
+        with Session(self.engine) as connection:
+            yield connection
+
+    def disconnect(self):
+        self.engine.dispose()

src/db/transaction.py

@@ -0,0 +1,58 @@
+from datetime import datetime, timedelta
+from typing import AsyncIterator, Iterable, List
+
+from rusty_results import Empty, Option, Some
+from sqlalchemy import Result
+from sqlmodel import select
+
+from db.clients import DbClient
+from node.models.transactions import Transaction
+from utils.datetime import increment_datetime
+
+
+class TransactionRepository:
+    def __init__(self, client: DbClient):
+        self.client = client
+
+    async def create(self, transaction: Iterable[Transaction]) -> None:
+        with self.client.session() as session:
+            session.add_all(transaction)
+            session.commit()
+
+    async def get_latest(self, limit: int, descending: bool = True) -> List[Transaction]:
+        statement = select(Transaction).limit(limit)
+        if descending:
+            statement = statement.order_by(Transaction.timestamp.desc())
+        else:
+            statement = statement.order_by(Transaction.timestamp.asc())
+
+        with self.client.session() as session:
+            results: Result[Transaction] = session.exec(statement)
+            return results.all()
+
+    async def updates_stream(self, timestamp_from: datetime) -> AsyncIterator[List[Transaction]]:
+        _timestamp_from = timestamp_from
+        while True:
+            statement = (
+                select(Transaction)
+                .where(Transaction.timestamp >= _timestamp_from)
+                .order_by(Transaction.timestamp.asc())
+            )
+
+            with self.client.session() as session:
+                transactions: List[Transaction] = session.exec(statement).all()
+
+            if len(transactions) > 0:
+                # POC: Assumes transactions are inserted in order and with a minimum 1 of second difference
+                _timestamp_from = increment_datetime(transactions[-1].timestamp)
+
+            yield transactions
+
+    async def get_earliest(self) -> Option[Transaction]:
+        with self.client.session() as session:
+            statement = select(Transaction).order_by(Transaction.slot.asc()).limit(1)
+            results: Result[Transaction] = session.exec(statement)
+            if (transaction := results.first()) is not None:
+                return Some(transaction)
+            else:
+                return Empty()

src/frontend/__init__.py

@@ -0,0 +1 @@
+from .mount import create_frontend_router

src/frontend/mount.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter
+from fastapi.staticfiles import StaticFiles
+from starlette.responses import FileResponse, Response
+
+from src import DIR_REPO
+
+STATIC_DIR = DIR_REPO.joinpath("static")
+INDEX_FILE = STATIC_DIR.joinpath("index.html")
+
+
+def index() -> Response:
+    return FileResponse(INDEX_FILE)
+
+
+def create_frontend_router() -> APIRouter:
+    router = APIRouter()
+
+    router.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+    router.get("/", include_in_schema=False)(index)
+
+    return router

src/lifespan.py

@@ -0,0 +1,14 @@
+import typing
+from contextlib import AsyncExitStack, asynccontextmanager
+
+from node.lifespan import node_lifespan
+
+if typing.TYPE_CHECKING:
+    from core.app import NBE
+
+
+@asynccontextmanager
+async def lifespan(app: "NBE"):
+    async with AsyncExitStack() as stack:
+        await stack.enter_async_context(node_lifespan(app))
+        yield

src/logs.py

@@ -0,0 +1,62 @@
+import os
+from logging.config import dictConfig
+
+_LEVEL = os.getenv("NBE_LOG_LEVEL", "INFO").upper()
+_SQLA_LEVEL = os.getenv("SQLALCHEMY_LOG_LEVEL", "ERROR").upper()
+
+_LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": True,
+    "formatters": {
+        "standard": {
+            "format": "[%(asctime)s] [%(levelname)s] %(name)s: %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+        "uvicorn": {
+            "format": "[%(asctime)s] [%(levelname)s] [uvicorn] %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+        "uvicorn_access": {
+            "format": '%(client_addr)s - "%(request_line)s" %(status_code)s',
+        },
+    },
+    "handlers": {
+        "console": {
+            "class": "logging.StreamHandler",
+            "level": "DEBUG",
+            "formatter": "standard",
+        },
+        "uvicorn": {
+            "class": "logging.StreamHandler",
+            "level": "DEBUG",
+            "formatter": "uvicorn",
+        },
+        "uvicorn_access": {
+            "class": "logging.StreamHandler",
+            "level": "WARNING",
+            "formatter": "uvicorn_access",
+        },
+    },
+    "root": {
+        "handlers": ["console"],
+        "level": _LEVEL,
+    },
+    "loggers": {
+        # ---- SQLAlchemy / SQLModel ----
+        "sqlalchemy": {"level": _SQLA_LEVEL, "handlers": [], "propagate": False},
+        "sqlalchemy.engine": {"level": _SQLA_LEVEL, "handlers": [], "propagate": False},
+        "sqlalchemy.pool": {"level": _SQLA_LEVEL, "handlers": [], "propagate": False},
+        "sqlalchemy.orm": {"level": _SQLA_LEVEL, "handlers": [], "propagate": False},
+        "sqlalchemy.dialects": {"level": _SQLA_LEVEL, "handlers": [], "propagate": False},
+        # ---- Uvicorn / FastAPI ----
+        "uvicorn": {"level": "INFO", "handlers": ["uvicorn"], "propagate": False},
+        # "uvicorn.error":  {"level": "INFO", "handlers": ["uvicorn"], "propagate": False},
+        "uvicorn.access": {"level": "WARNING", "handlers": ["uvicorn_access"], "propagate": False},
+        # Your app namespace
+        "app": {"level": _LEVEL, "handlers": ["console"], "propagate": False},
+    },
+}
+
+
+def setup_logging():
+    dictConfig(_LOGGING_CONFIG)

src/main.py

@@ -0,0 +1,35 @@
+import asyncio
+
+import uvicorn
+
+from app import create_app
+from logs import setup_logging
+
+
+async def main():
+    app = create_app()
+    config = uvicorn.Config(
+        app,
+        host="127.0.0.1",
+        port=8000,
+        reload=False,
+        loop="asyncio",
+        log_config=None,
+    )
+    server = uvicorn.Server(config)
+
+    try:
+        await server.serve()
+    except KeyboardInterrupt:
+        # Swallow debugger’s SIGINT
+        pass
+
+
+# Pycharm-Debuggable Uvicorn Server
+if __name__ == "__main__":
+    try:
+        setup_logging()
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        # Graceful stop triggered by debugger/CTRL-C
+        pass

src/node/api/base.py

@@ -0,0 +1,20 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+from node.models.blocks import Block
+from node.models.health import Health
+from node.models.transactions import Transaction
+
+
+class NodeApi(ABC):
+    @abstractmethod
+    async def get_health_check(self) -> Health:
+        pass
+
+    @abstractmethod
+    async def get_transactions(self) -> List[Transaction]:
+        pass
+
+    @abstractmethod
+    async def get_blocks(self) -> List[Block]:
+        pass

src/node/api/fake.py

@@ -0,0 +1,27 @@
+from random import choices, random
+from typing import List
+
+from node.api.base import NodeApi
+from node.models.blocks import Block
+from node.models.health import Health
+from node.models.transactions import Transaction
+
+
+def get_weighted_amount() -> int:
+    items = [1, 2, 3]
+    weights = [0.6, 0.3, 0.1]
+    return choices(items, weights=weights, k=1)[0]
+
+
+class FakeNodeApi(NodeApi):
+    async def get_health_check(self) -> Health:
+        if random() < 0.1:
+            return Health.from_unhealthy()
+        else:
+            return Health.from_healthy()
+
+    async def get_transactions(self) -> List[Transaction]:
+        return [Transaction.from_random() for _ in range(get_weighted_amount())]
+
+    async def get_blocks(self) -> List[Block]:
+        return [Block.from_random() for _ in range(1)]

src/node/api/http.py

@@ -0,0 +1,35 @@
+from urllib.parse import urljoin
+
+import requests
+
+from node.api.base import NodeApi
+
+
+class HttpNodeApi(NodeApi):
+    ENDPOINT_MANTLE_STATUS = "/mantle/status"
+    ENDPOINT_MANTLE_TRANSACTIONS = "/mantle/transactions"
+    ENDPOINT_MANTLE_BLOCKS = "/mantle/blocks"
+
+    def __init__(self, host: str, port: int, protocol: str = "http"):
+        self.protocol: str = protocol
+        self.host: str = host
+        self.port: int = port
+
+    @property
+    def base_url(self):
+        return f"{self.protocol}://{self.host}:{self.port}"
+
+    async def get_health_check(self) -> dict:
+        url = urljoin(self.base_url, self.ENDPOINT_MANTLE_STATUS)
+        response = requests.get(url, timeout=60)
+        return response.json()
+
+    async def get_transactions(self) -> list:
+        url = urljoin(self.base_url, self.ENDPOINT_MANTLE_TRANSACTIONS)
+        response = requests.get(url, timeout=60)
+        return response.json()
+
+    async def get_blocks(self) -> list:
+        url = urljoin(self.base_url, self.ENDPOINT_MANTLE_BLOCKS)
+        response = requests.get(url, timeout=60)
+        return response.json()

src/node/lifespan.py

@@ -0,0 +1,151 @@
+import random
+from asyncio import TaskGroup, create_task, sleep
+from contextlib import asynccontextmanager
+from itertools import chain
+from typing import TYPE_CHECKING, AsyncGenerator
+
+from rusty_results import Option
+
+from db.blocks import BlockRepository
+from db.clients import SqliteClient
+from db.transaction import TransactionRepository
+from node.api.fake import FakeNodeApi
+from node.api.http import HttpNodeApi
+from node.manager.docker import DockerModeManager
+from node.manager.fake import FakeNodeManager
+from node.models.blocks import Block
+from node.models.transactions import Transaction
+
+if TYPE_CHECKING:
+    from core.app import NBE
+
+
+@asynccontextmanager
+async def node_lifespan(app: "NBE") -> AsyncGenerator[None]:
+    # app.state.node_manager = DockerModeManager()
+    # app.state.node_api = HttpApi(host="127.0.0.1", port=3000)
+
+    db_client = SqliteClient()
+
+    app.state.node_manager = FakeNodeManager()
+    app.state.node_api = FakeNodeApi()
+    app.state.db_client = db_client
+    app.state.block_repository = BlockRepository(db_client)
+    app.state.transaction_repository = TransactionRepository(db_client)
+    try:
+        print("Starting node...")
+        await app.state.node_manager.start()
+        print("Node started.")
+
+        app.state.subscription_to_updates_handle = create_task(subscription_to_updates(app))
+        app.state.backfill = create_task(backfill(app))
+
+        yield
+    finally:
+        print("Stopping node...")
+        await app.state.node_manager.stop()
+        print("Node stopped.")
+
+
+# ================
+# BACKFILLING
+# ================
+# Legend:
+# BT = Block and/or Transaction
+# Steps:
+# 1. Subscribe to new BT and store them in the database.
+# 2. Backfill gaps between the earliest received BT from subscription (step 1.) and the latest BT in the database.
+# 3. Backfill gaps between the earliest BT in the database and genesis BT (slot 0).
+# Assumptions:
+# - BT are always filled correctly.
+# - There's at most 1 gap in the BT sequence: From genesis to earliest received BT from subscription.
+# - Slots are populated fully or not at all (no partial slots).
+# Notes:
+# - Upsert always.
+
+# ================
+# Fake
+_SUBSCRIPTION_START_SLOT = 5  # Simplification for now.
+# ================
+
+
+async def subscription_to_updates(app: "NBE") -> None:
+    print("✅ Subscription to new blocks and transactions started.")
+    async with TaskGroup() as tg:
+        tg.create_task(subscribe_to_new_blocks(app))
+        tg.create_task(subscribe_to_new_transactions(app))
+    print("Subscription to new blocks and transactions finished.")
+
+
+async def subscribe_to_new_blocks(
+    app: "NBE", interval: int = 5, subscription_start_slot: int = _SUBSCRIPTION_START_SLOT
+):
+    while app.state.is_running:
+        try:
+            new_block: Block = Block.from_random(slot_start=subscription_start_slot, slot_end=subscription_start_slot)
+            print("> New Block")
+            await app.state.block_repository.create((new_block,))
+            subscription_start_slot += 1
+        except Exception as e:
+            print(f"Error while subscribing to new blocks: {e}")
+        finally:
+            await sleep(interval)
+
+
+async def subscribe_to_new_transactions(app: "NBE", interval: int = 5):
+    while app.state.is_running:
+        try:
+            new_transaction: Transaction = Transaction.from_random()
+            print("> New TX")
+            await app.state.transaction_repository.create((new_transaction,))
+        except Exception as e:
+            print(f"Error while subscribing to new transactions: {e}")
+        finally:
+            await sleep(interval)
+
+
+async def backfill(app: "NBE", delay: int = 3) -> None:
+    await sleep(delay)  # Wait for some data to be present: Simplification for now.s
+
+    print("Backfilling started.")
+    async with TaskGroup() as tg:
+        tg.create_task(backfill_blocks(app))
+        tg.create_task(backfill_transactions(app))
+    print("✅ Backfilling finished.")
+
+
+async def backfill_blocks(app: "NBE"):
+    # Assuming at most one gap. This will be either genesis block (no gap) or the earliest received block from subscription.
+    # If genesis, do nothing.
+    # If earliest received block from subscription, backfill.
+    print("Checking for block gaps to backfill...")
+    earliest_block: Option[Block] = await app.state.block_repository.get_earliest()
+    earliest_block: Block = earliest_block.expects("Subscription should have provided at least one block by now.")
+    earliest_block_slot = earliest_block.slot
+    if earliest_block_slot == 0:
+        print("No need to backfill blocks, genesis block already present.")
+        return
+
+    print(f"Backfilling blocks from slot {earliest_block_slot - 1} down to 0...")
+
+    def n_blocks():
+        return random.choices((1, 2, 3), (6, 3, 1))[0]
+
+    blocks = (
+        (Block.from_random(slot_start=slot_index, slot_end=slot_index) for _ in range(n_blocks()))
+        for slot_index in reversed(range(0, earliest_block_slot))
+    )
+    flattened = list(chain.from_iterable(blocks))
+    await sleep(10)  # Simulate some backfilling delay
+    await app.state.block_repository.create(flattened)
+    print("Backfilling blocks completed.")
+
+
+async def backfill_transactions(app: "NBE"):
+    # Assume there's some TXs to backfill
+    n = random.randint(0, 5)
+    print(f"Backfilling {n} transactions...")
+    transactions = (Transaction.from_random() for _ in range(n))
+    await sleep(10)  # Simulate some backfilling delay
+    await app.state.transaction_repository.create(transactions)
+    print("Backfilling transactions completed.")

src/node/manager/base.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+
+
+class NodeManager(ABC):
+    @abstractmethod
+    async def start(self):
+        pass
+
+    @abstractmethod
+    async def stop(self):
+        pass

src/node/manager/docker.py

@@ -0,0 +1,28 @@
+from os import environ
+from pathlib import Path
+
+from python_on_whales.docker_client import DockerClient
+
+from node.manager.base import NodeManager
+
+
+class DockerModeManager(NodeManager):
+    COMPOSE_FILE: Path = Path(environ["NODE_COMPOSE_FILEPATH"])
+
+    def __init__(self):
+        self.client: DockerClient = DockerClient(
+            client_type="docker",
+            compose_files=[
+                self.COMPOSE_FILE,
+            ],
+        )
+
+    async def start(self):
+        self.client.compose.up(
+            detach=True,
+            build=False,
+            remove_orphans=True,
+        )
+
+    async def stop(self):
+        self.client.compose.down(remove_orphans=True, volumes=True)

src/node/manager/fake.py

@@ -0,0 +1,9 @@
+from node.manager.base import NodeManager
+
+
+class FakeNodeManager(NodeManager):
+    async def start(self):
+        pass
+
+    async def stop(self):
+        pass

src/node/models/blocks.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+
+from sqlmodel import Field
+
+from core.models import IdNbeModel
+from utils.random import random_hash
+
+
+class Block(IdNbeModel, table=True):
+    __tablename__ = "blocks"
+
+    slot: int
+    hash: str
+    parent_hash: str
+    transaction_count: int
+    timestamp: datetime = Field(default=None, index=True)
+
+    @classmethod
+    def from_random(cls, slot_start=1, slot_end=10_000) -> "Block":
+        import random
+
+        return cls(
+            slot=random.randint(slot_start, slot_end),
+            hash=random_hash(),
+            parent_hash=random_hash(),
+            transaction_count=random.randint(0, 500),
+            timestamp=datetime.now(),
+        )

src/node/models/health.py

@@ -0,0 +1,13 @@
+from core.models import IdNbeModel
+
+
+class Health(IdNbeModel):
+    healthy: bool
+
+    @classmethod
+    def from_healthy(cls) -> "Health":
+        return cls(healthy=True)
+
+    @classmethod
+    def from_unhealthy(cls) -> "Health":
+        return cls(healthy=False)

src/node/models/transactions.py

@@ -0,0 +1,30 @@
+import random
+from datetime import datetime
+from typing import Optional
+
+from sqlmodel import Field
+
+from core.models import IdNbeModel
+from utils.random import random_address, random_hash
+
+
+class Transaction(IdNbeModel, table=True):
+    __tablename__ = "transactions"
+
+    hash: str
+    block_hash: Optional[str] = Field(default=None, index=True)
+    sender: str
+    recipient: str
+    amount: float
+    timestamp: datetime = Field(default=None, index=True)
+
+    @classmethod
+    def from_random(cls) -> "Transaction":
+        return Transaction(
+            hash=random_hash(),
+            block_hash=random_hash(),
+            sender=random_address(),
+            recipient=random_address(),
+            amount=round(random.uniform(0.0001, 100.0), 6),
+            timestamp=datetime.now(),
+        )

src/router.py

@@ -0,0 +1,35 @@
+from functools import partial, update_wrapper
+from os import environ
+from urllib.request import Request
+
+from fastapi import APIRouter
+from starlette.responses import JSONResponse, Response
+
+from api.router import create_api_router
+from frontend import create_frontend_router
+
+
+async def _debug_router(_request: Request, *_, router: APIRouter) -> Response:
+    content = [
+        {
+            "path": route.path,
+            "name": route.name,
+            "methods": list(route.methods),
+        }
+        for route in router.routes
+    ]
+    return JSONResponse(content)
+
+
+def create_router() -> APIRouter:
+    router = APIRouter()
+
+    debug_router = partial(_debug_router, router=router)
+    update_wrapper(debug_router, _debug_router)
+
+    router.include_router(create_api_router(), prefix="/api")
+    if bool(environ.get("DEBUG")):
+        router.add_route("/debug", debug_router)
+    router.include_router(create_frontend_router())
+
+    return router

src/utils/datetime.py

@@ -0,0 +1,8 @@
+from datetime import datetime, timedelta
+
+
+# Increments a timestamp by the smallest possible unit (1 microsecond), in terms of DB precision.
+# This is used to avoid returning the same record again when querying for updates.
+# FIXME: Hardcoded
+def increment_datetime(timestamp: datetime) -> datetime:
+    return timestamp + timedelta(microseconds=1)

src/utils/random.py

@@ -0,0 +1,13 @@
+import random
+
+
+def random_hex(length: int) -> str:
+    return f"0x{random.getrandbits(length * 4):0{length}x}"
+
+
+def random_hash() -> str:
+    return random_hex(64)
+
+
+def random_address() -> str:
+    return random_hex(40)

static/index.html

@@ -0,0 +1,258 @@
+<!doctype html>
+<html lang="en">
+<head>
+    <meta charset="utf-8" />
+    <title>Nomos Block Explorer</title>
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <style>
+        :root {
+            --bg:#0b0e14; --card:#131722; --fg:#e6edf3; --muted:#9aa4ad;
+            --accent:#3fb950; --warn:#ffb86b;
+        }
+        * { box-sizing: border-box; }
+        body { margin:0; font:14px/1.4 system-ui,sans-serif; background:var(--bg); color:var(--fg); }
+        header { display:flex; gap:12px; align-items:center; padding:14px 16px; background:#0e1320; position: sticky; top:0; }
+        h1 { font-size:16px; margin:0; }
+        .pill { padding:4px 8px; border-radius:999px; background:#1b2133; color:var(--muted); font-size:12px; }
+        .pill.online { background: rgba(63,185,80,.15); color: var(--accent); }
+        .pill.offline { background: rgba(255,184,107,.15); color: var(--warn); }
+        .flash { animation: flash 700ms ease-out; }
+        @keyframes flash { from { box-shadow:0 0 0 0 rgba(63,185,80,.9) } to { box-shadow:0 0 0 14px rgba(63,185,80,0) } }
+        main { max-width: 1400px; margin: 20px auto; padding: 0 16px 40px; }
+        .card { background: var(--card); border: 1px solid #20263a; border-radius: 10px; overflow: hidden; }
+        .card-header { display:flex; justify-content:space-between; align-items:center; padding:12px 14px; border-bottom: 1px solid #1f2435; }
+
+        /* SCROLLER */
+        .table-wrapper {
+            overflow-y: auto;
+            overflow-x: auto;         /* horizontal scrolling back */
+            -webkit-overflow-scrolling: touch;
+            max-height: 60vh;
+            scrollbar-gutter: stable both-edges;
+            padding-right: 8px;
+        }
+
+        table {
+            border-collapse: collapse;
+            table-layout: fixed;
+            font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+            width: 100%;
+        }
+        /* Force tables to be wider than the container when needed so horizontal scroll appears */
+        .table-wrapper .table--blocks { min-width: 560px; }
+        .table-wrapper .table--txs    { min-width: 980px; } /* Hash + From→To + Amount + Time */
+
+        th, td {
+            text-align:left;
+            padding:8px 10px;
+            border-bottom:1px solid #1f2435;
+            vertical-align: top;
+            white-space: nowrap;
+            overflow: hidden;
+            text-overflow: ellipsis;
+        }
+        th { color: var(--muted); font-weight: normal; font-size: 13px; position: sticky; top: 0; background: var(--card); z-index: 1; }
+        tbody td { height: 28px; }
+        tr:nth-child(odd) { background: #121728; }
+
+        .twocol { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; margin-top: 16px; }
+        @media (max-width: 960px) { .twocol { grid-template-columns: 1fr; } }
+
+        .table--txs th:last-child, .table--txs td:last-child { padding-right: 16px; }
+
+        .mono { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; }
+        .amount { font-variant-numeric: tabular-nums; }
+    </style>
+</head>
+<body>
+<header>
+    <h1>Nomos Block Explorer</h1>
+    <span id="status-pill" class="pill">Connecting…</span>
+</header>
+
+<main>
+    <section class="twocol">
+        <!-- Blocks -->
+        <div class="card">
+            <div class="card-header">
+                <div><strong>Blocks</strong> <span class="pill" id="blocks-count">0</span></div>
+                <div style="color:var(--muted); font-size:12px;">/api/v1/blocks/stream</div>
+            </div>
+            <div class="table-wrapper">
+                <table class="table--blocks">
+                    <colgroup>
+                        <col style="width:100px" />
+                        <col style="width:260px" />
+                        <col style="width:80px" />
+                    </colgroup>
+                    <thead>
+                    <tr>
+                        <th>Slot</th>
+                        <th>Hash</th>
+                        <th>Txs</th>
+                    </tr>
+                    </thead>
+                    <tbody id="blocks-body"></tbody>
+                </table>
+            </div>
+        </div>
+
+        <!-- Transactions -->
+        <div class="card">
+            <div class="card-header">
+                <div><strong>Transactions</strong> <span class="pill" id="txs-count">0</span></div>
+                <div style="color:var(--muted); font-size:12px;">/api/v1/transactions/stream</div>
+            </div>
+            <div class="table-wrapper">
+                <table class="table--txs">
+                    <colgroup>
+                        <col style="width:260px" />
+                        <col /> <!-- From → To -->
+                        <col style="width:120px" />
+                        <col style="width:180px" /> <!-- Time -->
+                    </colgroup>
+                    <thead>
+                    <tr>
+                        <th>Hash</th>
+                        <th>From → To</th>
+                        <th>Amount</th>
+                        <th>Time</th>
+                    </tr>
+                    </thead>
+                    <tbody id="txs-body"></tbody>
+                </table>
+            </div>
+        </div>
+    </section>
+</main>
+
+<script>
+    const API_PREFIX = "/api/v1";
+    const HEALTH_ENDPOINT = `${API_PREFIX}/health/stream`;
+    const BLOCKS_ENDPOINT = `${API_PREFIX}/blocks/stream`;
+    const TXS_ENDPOINT = `${API_PREFIX}/transactions/stream`;
+    const TABLE_SIZE = 10;
+
+    // ---- Health pill ----
+    const pill = document.getElementById("status-pill");
+    let healthController = null;
+    let state = "connecting";
+    function setState(next) {
+        if (next === state) return;
+        state = next;
+        pill.className = "pill";
+        if (state === "online") { pill.textContent = "Online"; pill.classList.add("online","flash"); }
+        else if (state === "offline") { pill.textContent = "Offline"; pill.classList.add("offline","flash"); }
+        else { pill.textContent = "Connecting…"; }
+        setTimeout(() => pill.classList.remove("flash"), 750);
+    }
+    function applyHealth(obj) { if (typeof obj?.healthy === "boolean") setState(obj.healthy ? "online" : "offline"); }
+    async function connectHealth() {
+        if (healthController) healthController.abort();
+        healthController = new AbortController();
+        setState("connecting");
+        try {
+            const res = await fetch(HEALTH_ENDPOINT, { signal: healthController.signal, cache: "no-cache" });
+            if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
+            const reader = res.body.getReader(); const decoder = new TextDecoder(); let buf = "";
+            while (true) {
+                const { value, done } = await reader.read(); if (done) break;
+                buf += decoder.decode(value, { stream: true });
+                const lines = buf.split("\n"); buf = lines.pop() ?? "";
+                for (const line of lines) { if (!line.trim()) continue; try { applyHealth(JSON.parse(line)); } catch {} }
+            }
+        } catch (e) { if (!healthController?.signal.aborted) setState("offline"); }
+    }
+    connectHealth();
+
+    // ---- NDJSON reader ----
+    async function streamNDJSON(url, onItem, { signal } = {}) {
+        const res = await fetch(url, { headers: { "accept": "application/x-ndjson" }, signal, cache: "no-cache" });
+        if (!res.ok || !res.body) throw new Error(`Stream failed: ${res.status}`);
+        const reader = res.body.getReader(); const decoder = new TextDecoder(); let buf = "";
+        while (true) {
+            const { value, done } = await reader.read(); if (done) break;
+            buf += decoder.decode(value, { stream: true });
+            let idx;
+            while ((idx = buf.indexOf("\n")) >= 0) {
+                const line = buf.slice(0, idx).trim(); buf = buf.slice(idx + 1);
+                if (!line) continue; try { onItem(JSON.parse(line)); } catch {}
+            }
+        }
+        const last = buf.trim(); if (last) { try { onItem(JSON.parse(last)); } catch {} }
+    }
+
+    // ---- Table helpers ----
+    function ensureSize(tbody, colCount) {
+        while (tbody.rows.length < TABLE_SIZE) {
+            const row = document.createElement("tr");
+            for (let i = 0; i < colCount; i++) {
+                const td = document.createElement("td");
+                td.innerHTML = "&nbsp;";
+                row.appendChild(td);
+            }
+            tbody.appendChild(row);
+        }
+    }
+    function appendRow(tbody, cells) {
+        const tr = document.createElement("tr");
+        for (const cell of cells) {
+            const td = document.createElement("td");
+            if (cell && cell.html !== undefined) td.innerHTML = cell.html;
+            else td.textContent = String(cell ?? "");
+            if (cell && cell.class) td.className = cell.class;
+            tr.appendChild(td);
+        }
+        tbody.insertBefore(tr, tbody.firstChild);
+        while (tbody.rows.length > TABLE_SIZE) tbody.deleteRow(-1);
+    }
+    function shortHex(s, left = 6, right = 4) { if (!s) return ""; return s.length <= left + right + 2 ? s : `${s.slice(0,left)}…${s.slice(-right)}`; }
+
+    // Flexible timestamp formatter
+    function fmtTime(ts) {
+        if (ts == null) return "";
+        let d;
+        if (typeof ts === "number") d = new Date(ts < 1e12 ? ts * 1000 : ts);
+        else d = new Date(ts);
+        if (isNaN(d)) return "";
+        return d.toLocaleString(undefined, {
+            year: "numeric", month: "2-digit", day: "2-digit",
+            hour: "2-digit", minute: "2-digit", second: "2-digit"
+        });
+    }
+
+    // ---- Blocks ----
+    (function initBlocks() {
+        const body = document.getElementById("blocks-body");
+        const counter = document.getElementById("blocks-count");
+        let n = 0;
+        ensureSize(body, 3);
+        streamNDJSON(BLOCKS_ENDPOINT, (b) => {
+            appendRow(body, [
+                { html: `<span class="mono">${b.slot}</span>` },
+                { html: `<span class="mono" title="${b.hash}">${shortHex(b.hash, 10, 8)}</span>` },
+                b.transaction_count,
+            ]);
+            counter.textContent = (++n).toString();
+        }).catch(err => console.error("Blocks stream error:", err));
+    })();
+
+    // ---- Transactions ----
+    (function initTxs() {
+        const body = document.getElementById("txs-body");
+        const counter = document.getElementById("txs-count");
+        let n = 0;
+        ensureSize(body, 4); // 4 columns now
+        streamNDJSON(TXS_ENDPOINT, (t) => {
+            appendRow(body, [
+                { html: `<span class="mono" title="${t.hash}">${shortHex(t.hash, 10, 8)}</span>` },
+                { html: `<span class="mono" title="${t.sender}">${shortHex(t.sender)}</span> → <span class="mono" title="${t.recipient}">${shortHex(t.recipient)}</span>` },
+                { html: `<span class="amount">${Number(t.amount).toLocaleString(undefined, { maximumFractionDigits: 8 })}</span>` },
+                { html: `<span class="mono" title="${t.timestamp}">${fmtTime(t.timestamp)}</span>` },
+            ]);
+            counter.textContent = (++n).toString();
+        }).catch(err => console.error("Tx stream error:", err));
+    })();
+</script>
+</body>
+</html>