social-app/backend/src/core/config/settings.py

from __future__ import annotations

from pathlib import Path
from typing import ClassVar, Literal
from urllib.parse import quote

from pydantic import BaseModel, Field, computed_field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class RuntimeSettings(BaseModel):
    environment: Literal["dev", "test", "prod"] = "dev"
    service_name: str = "app"
    debug: bool = True
    log_level: str = "INFO"
    log_json: bool = True
    log_rotation: Literal["time", "size", "none"] = "time"
    log_rotation_when: str = "midnight"
    log_rotation_interval: int = 1
    log_rotation_backup_count: int = 14
    log_rotation_max_bytes: int = 10_000_000
    log_dir: str = "logs"
    log_error_dir: str = "logs/errors"
    log_file_name: str = "app.log"
    log_error_file_name: str = "error.log"
    log_sensitive_fields: list[str] = Field(
        default_factory=lambda: [
            "password",
            "secret",
            "token",
            "api_key",
            "authorization",
            "cookie",
            "client_ip",
            "user_id",
        ]
    )
    sql_log_queries: bool = False

    @field_validator("log_dir", mode="before")
    @classmethod
    def lock_log_dir(cls, _: object) -> str:
        return "logs"

    @field_validator("log_error_dir", mode="before")
    @classmethod
    def lock_log_error_dir(cls, _: object) -> str:
        return "logs/errors"

    @field_validator("log_file_name", mode="before")
    @classmethod
    def lock_log_file_name(cls, _: object) -> str:
        return "app.log"

    @field_validator("log_error_file_name", mode="before")
    @classmethod
    def lock_log_error_file_name(cls, _: object) -> str:
        return "error.log"


class CelerySettings(BaseModel):
    broker_url: str | None = None
    result_backend: str | None = None
    task_serializer: str = "json"
    result_serializer: str = "json"
    accept_content: list[str] = Field(default_factory=lambda: ["json"])
    timezone: str = "UTC"
    enable_utc: bool = True
    task_track_started: bool = True
    task_time_limit: int = 300
    task_soft_time_limit: int = 240
    task_default_retry_delay: int = 30
    task_max_retries: int = 3


class WebSettings(BaseModel):
    server: Literal["uvicorn", "gunicorn"] = "gunicorn"
    host: str = "0.0.0.0"
    port: int = Field(default=8000, ge=1, le=65535)
    reload: bool = False
    workers: int = Field(default=2, ge=1, le=64)
    worker_class: str = "uvicorn.workers.UvicornWorker"
    timeout: int = Field(default=60, ge=1, le=600)
    keepalive: int = Field(default=5, ge=1, le=120)
    log_level: Literal["debug", "info", "warning", "error", "critical"] = "info"


class GunicornSettings(BaseModel):
    enabled_in_prod: bool = True
    workers: int = 2
    worker_class: str = "uvicorn.workers.UvicornWorker"
    worker_connections: int = 1000
    timeout: int = 60
    graceful_timeout: int = 30
    keepalive: int = 5
    max_requests: int = 1000
    max_requests_jitter: int = 50
    preload_app: bool = False


class WorkerGroupSettings(BaseModel):
    concurrency: int = Field(default=2, ge=1, le=32)
    pool: Literal["prefork", "threads", "solo", "eventlet", "gevent"] = "prefork"
    time_limit: int = Field(default=300, ge=1, le=7200)
    soft_time_limit: int = Field(default=240, ge=1, le=3600)
    max_tasks_per_child: int = Field(default=200, ge=1, le=1000)
    prefetch_multiplier: int = Field(default=1, ge=1, le=10)


class WorkerSettings(BaseModel):
    groups: dict[str, WorkerGroupSettings] = Field(
        default_factory=lambda: {
            "critical": WorkerGroupSettings(
                concurrency=2,
                prefetch_multiplier=1,
                time_limit=300,
            ),
            "default": WorkerGroupSettings(
                concurrency=2,
                prefetch_multiplier=4,
                time_limit=600,
            ),
            "bulk": WorkerGroupSettings(
                concurrency=1,
                prefetch_multiplier=1,
                time_limit=3600,
                max_tasks_per_child=100,
            ),
        }
    )

    def get_group_config(self, group_name: str) -> WorkerGroupSettings:
        return self.groups.get(group_name, WorkerGroupSettings())


class CorsSettings(BaseModel):
    allow_origins: list[str] = Field(
        default_factory=lambda: [
            "http://localhost",
            "http://localhost:3000",
        ]
    )
    allow_credentials: bool = True
    allow_methods: list[str] = Field(default_factory=lambda: ["*"])
    allow_headers: list[str] = Field(default_factory=lambda: ["*"])


class RedisSettings(BaseModel):
    host: str = "redis"
    port: int = 6379
    password: str | None = None
    db: int = 0
    socket_connect_timeout: float = 1.0
    socket_timeout: float = 1.0
    max_connections: int = 10

    @computed_field
    @property
    def url(self) -> str:
        if self.password:
            password = quote(self.password, safe="")
            return f"redis://:{password}@{self.host}:{self.port}/{self.db}"
        return f"redis://{self.host}:{self.port}/{self.db}"


class SupabaseSettings(BaseModel):
    public_scheme: str = "http"
    public_host: str = "localhost"
    kong_http_port: int = 8000
    anon_key: str = "CHANGE_ME"
    service_role_key: str = "CHANGE_ME"
    jwt_secret: str | None = None

    @computed_field
    @property
    def public_url(self) -> str:
        return f"{self.public_scheme}://{self.public_host}:{self.kong_http_port}"

    @computed_field
    @property
    def api_external_url(self) -> str:
        return self.public_url

    @computed_field
    @property
    def url(self) -> str:
        return self.public_url


class DatabaseSettings(BaseModel):
    host: str = "localhost"
    port: int = 5432
    name: str = "postgres"
    user: str = "postgres"
    password: str = "CHANGE_ME"

    @computed_field
    @property
    def url(self) -> str:
        password = quote(self.password, safe="")
        return (
            f"postgresql+asyncpg://{self.user}:{password}"
            f"@{self.host}:{self.port}/{self.name}"
        )


def _resolve_env_file() -> str:
    current = Path(__file__).resolve()
    for parent in [current, *current.parents]:
        candidate = parent / ".env"
        if candidate.is_file():
            return str(candidate)
    return ".env"


class Settings(BaseSettings):
    runtime: RuntimeSettings = RuntimeSettings()
    web: WebSettings = WebSettings()
    gunicorn: GunicornSettings = GunicornSettings()
    cors: CorsSettings = CorsSettings()
    redis: RedisSettings = RedisSettings()
    supabase: SupabaseSettings = SupabaseSettings()
    celery: CelerySettings = CelerySettings()
    database: DatabaseSettings = DatabaseSettings()
    worker: WorkerSettings = WorkerSettings()

    @computed_field
    @property
    def database_url(self) -> str:
        return self.database.url

    @computed_field
    @property
    def celery_broker_url(self) -> str:
        return self.celery.broker_url or self.redis.url

    @computed_field
    @property
    def celery_result_backend(self) -> str:
        return self.celery.result_backend or self.redis.url

    model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
        env_file=_resolve_env_file(),
        env_prefix="SOCIAL_",
        env_nested_delimiter="__",
        case_sensitive=False,
        extra="ignore",
    )


config = Settings()