bootnode/api/bootnode/config.py

"""Application configuration."""

from functools import lru_cache
from typing import Annotated, Literal

from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict


class Settings(BaseSettings):
    """Application settings loaded from environment variables."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    # Application
    app_name: str = "Bootnode"
    app_env: Literal["development", "staging", "production"] = "development"
    debug: bool = False
    api_prefix: str = "/v1"

    # Database — Base/SQLite is canonical (VFS→SeaweedFS for durability);
    # PostgreSQL DSNs are still accepted for legacy/multi-instance. Plain str
    # (not PostgresDsn) so a `sqlite+aiosqlite:///…` URL validates.
    database_url: str = Field(
        default="sqlite+aiosqlite:////data/bootnode/bootnode.db"
    )
    db_pool_size: int = 20
    db_max_overflow: int = 10

    # Hanzo KV (RESP): cache, rate limits, usage counters, job queue
    kv_url: str = Field(default="redis://localhost:6379/0")

    # Authentication
    jwt_secret: str = Field(default="change-me-in-production")
    jwt_algorithm: str = "HS256"
    jwt_expire_minutes: int = 60 * 24  # 24 hours
    api_key_salt: str = Field(default="change-me-in-production")

    # Hanzo IAM Integration (hanzo.id, zoo.id, lux.id, pars.id)
    iam_url: str = "https://iam.hanzo.ai"
    iam_client_id: str = ""
    iam_client_secret: str = ""
    enable_multi_tenant: bool = True
    allowed_orgs: Annotated[list[str], NoDecode] = ["hanzo", "zoo", "lux", "pars"]
    frontend_url: str = "http://localhost:3001"
    allowed_origins: Annotated[list[str], NoDecode] = []  # Extra CORS origins

    @field_validator("allowed_orgs", "allowed_origins", mode="before")
    @classmethod
    def _split_csv(cls, v):
        """Read list settings as comma-separated env values.

        pydantic-settings decodes a ``list[str]`` field from the environment as
        JSON, so the plain ``a,b,c`` these variables are written as everywhere
        (Deployment manifests, .env, compose) raised SettingsError before any
        validator ran — constructing Settings() at all was fatal. NoDecode hands
        the raw string here instead.
        """
        if isinstance(v, str):
            return [item.strip() for item in v.split(",") if item.strip()]
        return v

    # Rate Limiting
    rate_limit_requests: int = 100  # per minute
    rate_limit_compute_units: int = 1000  # per minute

    # Chain RPCs (upstream nodes)
    eth_mainnet_rpc: str = ""
    eth_sepolia_rpc: str = ""
    eth_holesky_rpc: str = ""
    polygon_mainnet_rpc: str = ""
    polygon_amoy_rpc: str = ""
    arbitrum_mainnet_rpc: str = ""
    arbitrum_sepolia_rpc: str = ""
    optimism_mainnet_rpc: str = ""
    optimism_sepolia_rpc: str = ""
    base_mainnet_rpc: str = ""
    base_sepolia_rpc: str = ""
    avalanche_mainnet_rpc: str = ""
    avalanche_fuji_rpc: str = ""
    bsc_mainnet_rpc: str = ""
    bsc_testnet_rpc: str = ""
    lux_mainnet_rpc: str = ""
    lux_testnet_rpc: str = ""
    lux_devnet_rpc: str = ""
    solana_mainnet_rpc: str = ""
    solana_devnet_rpc: str = ""
    btc_mainnet_rpc: str = ""
    btc_testnet_rpc: str = ""
    btc_rpc_user: str = "bootnode"
    btc_rpc_password: str = ""

    # DataStore (OLAP analytics backend)
    datastore_url: str = "clickhouse://bootnode:bootnode@localhost:8123/bootnode"
    datastore_pool_size: int = 10

    # Vector Search (Qdrant)
    qdrant_url: str = "http://localhost:6333"
    qdrant_api_key: str = ""

    # Full-Text Search (Meilisearch)
    meilisearch_url: str = "http://localhost:7700"
    meilisearch_api_key: str = ""

    # Lux Indexer (Omnichain data)
    indexer_url: str = "http://localhost:5000"

    # Cloudflare (Edge Caching)
    cloudflare_zone_id: str = ""
    cloudflare_api_token: str = ""

    # External Services
    ipfs_gateway: str = "https://ipfs.io/ipfs/"

    # Webhooks
    webhook_timeout: int = 30
    webhook_max_retries: int = 5

    # Hanzo Commerce (Square-based billing)
    # Accepts both COMMERCE_URL and HANZO_COMMERCE_URL env vars for backward compat
    commerce_url: str = Field(
        default="https://commerce.hanzo.ai",
        validation_alias=AliasChoices("commerce_url", "hanzo_commerce_url"),
    )
    commerce_api_key: str = Field(
        default="",
        validation_alias=AliasChoices("commerce_api_key", "hanzo_commerce_api_key"),
    )
    commerce_webhook_secret: str = Field(
        default="",
        validation_alias=AliasChoices("commerce_webhook_secret", "hanzo_commerce_webhook_secret"),
    )

    # ERC-4337 Bundler
    bundler_private_key: str = ""
    bundler_beneficiary: str = ""

    # ZAP Server (native Cap'n Proto RPC)
    zap_enabled: bool = True
    zap_host: str = "0.0.0.0"
    zap_port: int = 9999

    # Hanzo O11y (OTEL)
    otel_enabled: bool = True
    otel_endpoint: str = "http://otel-collector.hanzo.svc:4317"
    otel_service_name: str = "bootnode-api"

    # Deployment Target
    deploy_target: Literal["docker", "process", "kubernetes"] = "docker"
    deploy_compose_file: str = "infra/compose.yml"
    deploy_k8s_namespace: str = "bootnode"
    deploy_k8s_context: str = ""

    # Helm / Lux Fleet
    helm_chart_path: str = "/opt/charts/lux"
    helm_binary: str = "helm"

    @property
    def is_production(self) -> bool:
        return self.app_env == "production"

    def validate_production(self) -> None:
        """Fail fast if production is missing critical secrets."""
        if not self.is_production:
            return
        insecure = []
        if self.jwt_secret == "change-me-in-production":
            insecure.append("JWT_SECRET")
        if self.api_key_salt == "change-me-in-production":
            insecure.append("API_KEY_SALT")
        if self.bundler_private_key == "":
            insecure.append("BUNDLER_PRIVATE_KEY")
        if insecure:
            raise RuntimeError(
                f"Production mode requires these secrets: {', '.join(insecure)}"
            )


@lru_cache
def get_settings() -> Settings:
    """Get cached settings instance."""
    return Settings()