import base64
import json
from pathlib import Path
from typing import Annotated

from pydantic import field_validator
from pydantic_settings import BaseSettings, NoDecode


def _default_env_files() -> tuple[str, ...]:
    current_file = Path(__file__).resolve()
    repo_root = current_file.parents[3]
    server_root = current_file.parents[2]
    return (
        str(repo_root / ".env"),
        str(server_root / ".env"),
    )


class Settings(BaseSettings):
    # App
    APP_ENV: str = "development"
    APP_DEBUG: bool = True
    APP_DOMAIN: str = "localhost"
    IDENTITY_SYNC_CA_BUNDLE_ROOT: str = "/etc/helix/ca-certificates"

    # Database
    DATABASE_URL: str  # required — startup will fail if not set in .env
    DATABASE_URL_SYNC: str  # required — startup will fail if not set in .env

    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"

    # ── Redis connection resilience (2026-08-13 incident) ─────────────
    # Four unrelated scheduled jobs (alicloud_intl.utilization,
    # data_access.bigquery_apply_reconcile, provisioning.dispatch,
    # onboarding.reconcile) failed inside 10 minutes with the same
    # ``TimeoutError: Timeout connecting to server`` raised out of the
    # single-flight lock acquire. The shared client was built with a bare
    # ``from_url(REDIS_URL, decode_responses=True)`` — no connect/read
    # deadline, no TCP keepalive and no liveness probe on checkout. Crons
    # are the worst case for that: the worker's pooled connections sit idle
    # for the whole gap between ticks, long enough for the network path
    # (ElastiCache idle reaping / NAT / SG state expiry) to drop them
    # silently, and the next tick checks out a dead socket. See
    # ``app/core/redis_client.py`` for how these are applied.
    #
    # Connect deadline. Bounds a black-holed TCP handshake: without it the
    # connect blocks on the OS-level SYN retry ceiling (~2 min on Linux)
    # while holding a job's concurrency slot, so one Redis blip drains the
    # gate for everyone. 5s is ~50× a healthy in-VPC connect (~2ms).
    REDIS_CONNECT_TIMEOUT_SECS: float = 5.0
    # Read deadline for a command's response. Every Helix Redis command is
    # O(1)-ish (SET NX / GET / EVAL of a 3-line script / TTL / EXISTS) and
    # nothing blocks server-side (no BLPOP / XREAD / SUBSCRIBE anywhere in
    # the codebase — checked), so 10s only ever fires on a wedged path.
    REDIS_SOCKET_TIMEOUT_SECS: float = 10.0
    # Liveness probe interval: redis-py PINGs a pooled connection idle
    # longer than this before handing it to a caller. Measured (2026-08-13,
    # against a real Redis with the connection killed server-side): on its
    # own this does NOT recover — the PING fails and the error surfaces
    # from there instead of from the caller's command. What it buys is
    # WHERE the failure lands: on a throwaway PING, before the caller's
    # command has been written to a half-dead socket. That is what makes
    # the retry below safe to take for any command rather than only
    # provably-idempotent ones. 30s is well under any deployed idle timeout
    # (NAT gateway 350s, SG connection tracking 350s).
    REDIS_HEALTH_CHECK_INTERVAL_SECS: int = 30
    # Command-level retry attempts on ConnectionError / TimeoutError. Each
    # retry disconnects and reconnects first. **This is the knob that
    # actually recovers** — same measurement: retry alone survives a
    # server-side-killed idle connection, the health check alone does not.
    # Worst case added latency during a real outage is bounded by
    # ``attempts × (connect timeout + backoff cap)``.
    REDIS_MAX_RETRIES: int = 3
    # Pool ceiling per process. redis-py defaults to effectively unbounded,
    # which turns a slow-Redis episode into an unbounded connection fan-out
    # against the server's ``maxclients``. 50 is far above steady-state
    # concurrency (SCHEDULED_JOB_MAX_CONCURRENCY=8 + request handlers).
    REDIS_MAX_CONNECTIONS: int = 50

    @field_validator(
        "REDIS_CONNECT_TIMEOUT_SECS",
        "REDIS_SOCKET_TIMEOUT_SECS",
    )
    @classmethod
    def _validate_redis_timeouts(cls, v: float) -> float:
        if not 0.5 <= v <= 120.0:
            raise ValueError(
                "Redis timeouts must be in [0.5, 120] seconds — a sub-second "
                "deadline turns normal jitter into spurious failures, and a "
                "2-minute one defeats the point of having a deadline."
            )
        return v

    @field_validator("REDIS_HEALTH_CHECK_INTERVAL_SECS")
    @classmethod
    def _validate_redis_health_check(cls, v: int) -> int:
        if not 0 <= v <= 3600:
            raise ValueError(
                "REDIS_HEALTH_CHECK_INTERVAL_SECS must be in [0, 3600] "
                "(0 disables the pre-checkout liveness PING)."
            )
        return v

    @field_validator("REDIS_MAX_RETRIES")
    @classmethod
    def _validate_redis_max_retries(cls, v: int) -> int:
        if not 0 <= v <= 10:
            raise ValueError("REDIS_MAX_RETRIES must be in [0, 10].")
        return v

    @field_validator("REDIS_MAX_CONNECTIONS")
    @classmethod
    def _validate_redis_max_connections(cls, v: int) -> int:
        if not 1 <= v <= 1000:
            raise ValueError("REDIS_MAX_CONNECTIONS must be in [1, 1000].")
        return v

    # Auth
    ALLOW_PASSWORD_LOGIN: bool = False

    # JWT
    JWT_SECRET_KEY: str  # required — startup will fail if not set in .env
    JWT_ALGORITHM: str = "HS256"
    # Access TTL bumped 30 → 240min (4h) 2026-06-02. Employees were
    # being silently logged out when a release-time backend restart
    # collided with the every-30-min refresh hop — 8× the refresh
    # frequency = 8× the surface for that race. Leaver-pending users
    # still ride the 5-min ``LEAVER_PENDING_ACCESS_TTL_MINUTES`` cap
    # in ``core/security.py``, so offboarding security is unchanged.
    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 240
    JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 7

    @field_validator("JWT_SECRET_KEY")
    @classmethod
    def validate_jwt_secret_strength(cls, v: str) -> str:
        """Pre-launch hardening 2026-05-04: enforce a minimum key
        strength on JWT_SECRET_KEY so an operator boot-up with a
        4-char value (HS256 brute-force in seconds) is rejected at
        process start instead of becoming a token-forgery vector.

        Accepts any string >= 32 chars OR any base64-decodable value
        whose decoded byte length is >= 32. Mirrors the
        CREDENTIAL_ENCRYPTION_KEY validator pattern.
        """
        if not v or not isinstance(v, str):
            raise ValueError("JWT_SECRET_KEY must be set")
        if len(v) >= 32:
            return v
        # Fallback: maybe operator pasted a base64url 32-byte key.
        try:
            decoded = base64.urlsafe_b64decode(v + "==")
            if len(decoded) >= 32:
                return v
        except Exception:
            pass
        raise ValueError(
            "JWT_SECRET_KEY too weak — must be >= 32 chars (or a "
            "base64url value decoding to >= 32 bytes). HS256 with a "
            "shorter key is brute-forceable."
        )

    # Google OAuth
    GOOGLE_CLIENT_ID: str  # required — startup will fail if not set in .env

    # SuccessFactors + Feishu config (base_url, credentials, webhook
    # secrets, approval codes, event tokens) lives in the
    # platform_configs table — managed via Admin > 配置中心. No env
    # vars for these integrations anymore.

    # Trusted reverse-proxy peers — only these may set X-Real-IP /
    # X-Forwarded-For. Entries may be single IPs or CIDR networks
    # (``172.16.0.0/12`` covers the docker-compose bridge nginx sits on
    # in production — the container IP is dynamic, so an exact-match
    # list can never pin it; see app/core/net.py). Defaults to loopback
    # only for bare-metal dev; production sets it via env/SSM.
    # ``NoDecode``: tell pydantic-settings NOT to JSON-decode this env
    # value at the source level. By default a ``list[str]`` field is
    # "complex", so the EnvSettingsSource runs ``json.loads`` on the raw
    # string BEFORE any ``mode="before"`` validator — and a value coming
    # from an ``env_file`` (``127.0.0.1,::1,…`` or a Compose-mangled
    # ``[a,b]``) is not valid JSON, so it raised ``SettingsError`` at
    # boot and broke the EC2 deploy migration step (2026-06-12). With
    # NoDecode the raw string is handed to ``parse_trusted_proxy_ips``
    # below, which accepts JSON / comma / mangled-bracket forms.
    TRUSTED_PROXY_IPS: Annotated[list[str], NoDecode] = ["127.0.0.1", "::1"]

    @field_validator("TRUSTED_PROXY_IPS", mode="before")
    @classmethod
    def parse_trusted_proxy_ips(cls, v: object) -> list[str]:
        """Accept a JSON array OR a comma-separated string from env.

        Paired with ``NoDecode`` on the field above (so this receives the
        raw env string, not a pre-decoded value). Try JSON first, else
        split on commas (stripping any stray brackets/quotes left by
        Docker Compose env-file quote mangling). render-env.sh writes the
        comma form for exactly this reason.
        """
        if isinstance(v, str):
            s = v.strip()
            try:
                parsed = json.loads(s)
                if isinstance(parsed, list):
                    return [str(x).strip() for x in parsed if str(x).strip()]
            except (json.JSONDecodeError, TypeError):
                pass
            s = s.strip("[]")
            return [
                part.strip().strip('"').strip("'")
                for part in s.split(",")
                if part.strip().strip('"').strip("'")
            ]
        return v  # type: ignore[return-value]

    # Login rate limiting (applied per source IP)
    LOGIN_RATE_LIMIT_MAX: int = 20       # max attempts per window
    LOGIN_RATE_LIMIT_WINDOW_SECONDS: int = 900  # 15 minutes

    # Refresh-token rate limiting — separate, wider bucket than
    # ``LOGIN_RATE_LIMIT_*``. Legit traffic on ``/auth/refresh`` is
    # driven by background polling + multi-tab + brief network blips,
    # and one IP can fan in many distinct users (corporate NAT, VPN
    # exits). When the two buckets were shared, a single user whose
    # access_token expired and reloaded a few times could burn the
    # ``/auth/google/verify`` quota and lock themselves out for the
    # full window. 2026-05-19 prod incident: separated.
    REFRESH_RATE_LIMIT_MAX: int = 200    # max /auth/refresh per IP per window
    REFRESH_RATE_LIMIT_WINDOW_SECONDS: int = 900  # 15 minutes

    # Sensitive endpoint rate limiting (PII reveal, API key reveal, export) — per user
    SENSITIVE_RATE_LIMIT_MAX: int = 10         # max reveal/export calls per user per window
    SENSITIVE_RATE_LIMIT_WINDOW_SECONDS: int = 3600  # 1 hour
    UPLOAD_PRESIGN_MAX_PER_HOUR: int = 60
    UPLOAD_PRESIGN_MAX_BYTES_PER_DAY: int = 200 * 1024 * 1024
    APPROVAL_UPLOAD_MAX_PER_HOUR: int = 60
    APPROVAL_UPLOAD_MAX_BYTES_PER_DAY: int = 1024 * 1024 * 1024
    UPLOAD_ORPHAN_RETENTION_HOURS: int = 24

    # AI copilot rate limiting — per user. Each /copilot/chat turn runs up to
    # several LLM round-trips against the org's DashScope/Bailian budget, so
    # an authenticated employee looping the endpoint could drive unbounded
    # spend. Generous enough for a real interactive chat session, bounded
    # enough to cap abuse. Separate from SENSITIVE_* (10/h would break chat).
    COPILOT_RATE_LIMIT_MAX: int = 60           # max copilot turns per user per window
    COPILOT_RATE_LIMIT_WINDOW_SECONDS: int = 3600  # 1 hour

    # AI Labs is a playground, so the budget is deliberately generous — the
    # point of the surface is that people use it. This limiter exists to
    # stop a script, not to ration an afternoon of real work: 240/hour is
    # one call every 15 seconds sustained for an hour, which no human
    # reaches by typing. The per-day *product* ceilings are separate,
    # operator-configurable and off by default (``ai_labs.quota.*``).
    AI_LABS_RATE_LIMIT_MAX: int = 240
    AI_LABS_RATE_LIMIT_WINDOW_SECONDS: int = 3600
    # Generation jobs cost far more per call than a chat turn and are the
    # only path that writes to S3, so they get their own, tighter budget.
    AI_LABS_GENERATION_RATE_LIMIT_MAX: int = 60
    AI_LABS_GENERATION_RATE_LIMIT_WINDOW_SECONDS: int = 3600

    # 安全 FAQ ask (/api/v1/me/security-faq/ask) — per user. Most asks are
    # answered by keyword search and cost nothing; only a zero-hit query
    # reaches the model. The cap exists for that tail, and is deliberately
    # separate from COPILOT_* so an employee who has been chatting with the
    # assistant can still look up a security answer.
    SECURITY_FAQ_ASK_RATE_LIMIT_MAX: int = 40
    SECURITY_FAQ_ASK_RATE_LIMIT_WINDOW_SECONDS: int = 3600  # 1 hour

    # Insights impact scorecard — per-process TTL cache for the heavy
    # all-time aggregation (~60 COUNTs incl. full-table audit_logs).
    # The projection board polls every 60 s, so the sensitive-endpoint
    # rate limiter (10/h) is the wrong tool; a short cache caps DB load
    # at one aggregation per process per window instead. 0 disables
    # (tests set this to 0 so seeded rows are visible immediately).
    INSIGHTS_IMPACT_CACHE_SECONDS: int = 60

    # 成本口径读面(财务瀑布 / methodology 对数区)的 Redis 响应缓存
    # TTL。这两个面按月分块聚合 cmdb_cost_facts(生产 ~100 万行/月,
    # 单请求 6-24 个分块查询、首载 5-25s)——缓存吸收重复加载。月度
    # 对账页容忍 30 分钟滞后。0 禁用(tests 置 0 保证种子行立即可见,
    # 镜像 INSIGHTS_IMPACT_CACHE_SECONDS 先例)。
    COST_READ_CACHE_TTL_SECONDS: int = 1800

    # MCP rate limiting — per PAT id (success path) and per IP (failed
    # auth path). 500 employees can each create their own PAT, so a
    # compromised PAT or a runaway LLM client must not be able to fill
    # the audit table or saturate DB connections.
    MCP_RATE_LIMIT_PER_PAT_MAX: int = 600         # ~10 req/s burst budget
    MCP_RATE_LIMIT_PER_PAT_WINDOW_SECONDS: int = 60
    MCP_AUTH_FAILURE_LIMIT_PER_IP_MAX: int = 30   # caps brute-force scan
    MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS: int = 60
    # Hard cap on the JSON-RPC request body. The MCP endpoint is the
    # only public surface FastAPI doesn't already filter at nginx in
    # production; an unbounded POST would let one PAT OOM the worker.
    MCP_REQUEST_BODY_MAX_BYTES: int = 1_048_576   # 1 MiB

    # MCP OAuth 2.1 (Claude / Cursor connector self-onboarding).
    #
    # ``PUBLIC_BASE_URL`` is the canonical https origin this deploy is
    # reached at (``https://helix.pingcap.net`` in prod). It anchors the
    # OAuth *issuer*, the discovery-metadata endpoint URLs, and the
    # audience the MCP access token is bound to. Leave empty to derive it
    # per-request from the ``Host`` + ``X-Forwarded-Proto`` headers — fine
    # for local dev / tests, but pin it in prod so the issuer is stable
    # regardless of which host header a client sends. No trailing slash.
    PUBLIC_BASE_URL: str = ""
    # Short access-token TTL (a leaked token's blast radius); the connector
    # silently refreshes. Refresh tokens rotate single-use on every use.
    MCP_OAUTH_ACCESS_TTL_MINUTES: int = 60
    MCP_OAUTH_REFRESH_TTL_DAYS: int = 30
    # Authorization-code lifetime (single-use; 60 s is ample for the
    # browser redirect → token exchange round-trip).
    MCP_OAUTH_CODE_TTL_SECONDS: int = 60

    @field_validator("PUBLIC_BASE_URL", mode="before")
    @classmethod
    def _strip_public_base_url(cls, v: object) -> str:
        if isinstance(v, str):
            return v.strip().rstrip("/")
        return ""

    # Minimum supported helix CLI version (release part X.Y.Z). A CLI
    # request whose ``User-Agent: helix-cli/<ver>`` is BELOW this floor
    # is refused at the MCP edge with 426 Upgrade Required — the
    # kill-switch for shipping a fix and cutting off known-buggy /
    # vulnerable old CLIs fleet-wide. "0.0.0" = no enforcement (every
    # build allowed). Bump via SSM + restart to lock out old clients.
    CLI_MIN_SUPPORTED_VERSION: str = "0.0.0"

    # Directory pull surface (identity consumers / SCIM) rate limiting.
    # Per-consumer success-path cap is overridable per row via
    # ``identity_consumers.rate_limit_per_minute``; this is the platform
    # default. Failed-auth windows mirror the MCP posture: stop token
    # scanners pre-DB and keep the audit table bounded.
    DIRECTORY_API_RATE_LIMIT_PER_MINUTE: int = 300
    DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_MAX: int = 30
    DIRECTORY_API_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS: int = 60
    # Keyset-feed visibility horizon: rows whose updated_at is younger
    # than this many seconds are held back from the directory list
    # endpoints so a transaction that commits late can't be skipped by
    # a cursor that already advanced past its timestamp. 0 disables.
    DIRECTORY_API_FEED_SAFETY_LAG_SECONDS: int = 5

    # HR sync hygiene — when True, SF employees whose ``email`` is a
    # known placeholder (``dummy*@...``, ``<user_id>@successfactors.local``,
    # the built-in SAP stubs) are skipped during preview instead of being
    # imported into ``identity_persons``. Leave False in test/staging
    # environments that intentionally ingest placeholder fixtures.
    HR_SKIP_PLACEHOLDER_EMAILS: bool = True

    # Comma-separated list of SF ``userId`` values that are service /
    # API admin accounts, not real employees. Skipped at preview time
    # alongside placeholder emails. Extend when new admin accounts show
    # up in SF (``APIadmin``, ``ONBadmin``, ``sfadmin`` cover PingCAP
    # 2026-04). Case-insensitive match.
    HR_SKIP_USER_IDS: str = (
        # SF service / admin / integration accounts that are NOT employees
        # and must never be ingested as people or provisioned a Google /
        # Feishu account. MokaAPI = the Moka ATS API service account;
        # TimeAdmin / LifecyleAdmin = SF admin/test accounts. Added
        # 2026-07-13 after all three were repeatedly picked up as "new
        # hires" by the onboarding provisioners (daily Google-mailbox
        # creation failures carded to the IT contact; would also mint junk
        # Feishu accounts via provision_feishu_account).
        "APIadmin,ONBadmin,sfadmin,admin,MokaAPI,TimeAdmin,LifecyleAdmin"
    )

    @field_validator("HR_SKIP_USER_IDS", mode="before")
    @classmethod
    def _normalize_skip_user_ids(cls, v: object) -> str:
        if isinstance(v, list):
            return ",".join(str(x) for x in v)
        return v  # type: ignore[return-value]

    # Comma-separated email domains that mark a row as non-employee.
    # Checked at TWO points in the SF lifecycle pipeline:
    #   (1) preview time on the SF-returned ``User.email`` — drops SAP
    #       implementation-partner accounts whose mailbox is on a
    #       vendor domain (``ECadmin@gaiaworks.cn``,
    #       ``COMPadmin@partner.gaiaworks.cn`` on PingCAP 2026-05);
    #   (2) upsert time on the email finally written to
    #       ``identity_persons.work_email`` — catches SF-side duplicate
    #       userIds whose desired email collided with the canonical
    #       row's, forcing ``_resolve_unique_work_email`` to fall back
    #       to ``<userId>@successfactors.local`` (PingCAP 2026-05:
    #       SF userIds 100113/80100057 are duplicates of P01115/P01134).
    # Subdomain match: ``partner.gaiaworks.cn`` matches ``gaiaworks.cn``
    # via ``endswith('.gaiaworks.cn')``. Case-insensitive.
    # ``sap.com`` added 2026-05-30: the SAP implementation tenant carries
    # a ``v4admin`` service account with ``dummy@sap.com`` that surfaces
    # in ``status='f'`` listings once you stop filtering by status — it
    # would have polluted the wangdi-PR forfeited-user backfill with a
    # noise row (``42667470E143D`` displayName, no real human behind it).
    # Same skip semantics as ``gaiaworks.cn`` (subdomain endswith match).
    HR_SKIP_EMAIL_DOMAINS: str = "gaiaworks.cn,successfactors.local,sap.com"

    @field_validator("HR_SKIP_EMAIL_DOMAINS", mode="before")
    @classmethod
    def _normalize_skip_email_domains(cls, v: object) -> str:
        if isinstance(v, list):
            return ",".join(str(x) for x in v)
        return v  # type: ignore[return-value]

    # SF EC Time-Off ``TimeAccountType.externalCode`` codes that map
    # onto the four ``identity_persons.annual_leave_*`` columns.
    # PingCAP CN tenant uses ``CHN_AnnualLeave`` (法定年假) and
    # ``CHN_PL`` (福利年假); EU / other tenants must override before
    # the annual-leave carryover sync runs. Empty value → that bucket
    # is left manual-only (NULL in DB, HR fills in 年假计算器 page).
    HR_TIMEOFF_LEGAL_ACCOUNT_TYPE: str = "CHN_AnnualLeave"
    HR_TIMEOFF_WELFARE_ACCOUNT_TYPE: str = "CHN_PL"

    # CORS
    CORS_ORIGINS: list[str] = ["http://localhost:3000"]

    # CMDB cloud sync — sweep guard tolerance. ``_safe_sweep`` (and the
    # mirrored scope/role sweeps in lifecycle_service) refuse to
    # soft-delete rows when ``upserted < baseline * (1 - tolerance/100)``,
    # i.e. when the cloud returned fewer rows than the previous
    # successful sync by more than this percentage. Default 20 (was
    # hardcoded 5 pre 2026-05-19): the 5% floor was tripping on
    # multi-region clouds where one flaky region returns 5–15% of the
    # fleet empty, leaving the sync wedged for hours waiting for the
    # cloud to fully recover. Raise this only after confirming the
    # gap is a real cloud-side fluctuation, not a driver regression.
    # Bounds: ``[0, 100]``. The semantics is "% of drop we will
    # tolerate before refusing to sweep":
    #   * 0   — STRICTEST: any drop trips the guard (upserted < baseline → refuse).
    #   * 20  — default: drops up to 20% are absorbed; >20% trips.
    #   * 100 — DISABLED (drop-ratio leg only — zero-rows guard
    #           remains unconditional): no drop ratio ever trips
    #           since the floor collapses to 0.
    # Promotion to runtime-dynamic (no restart): the platform_settings
    # row ``cmdb_sync.sweep_drop_tolerance_percent`` already takes
    # precedence over this value — this remains the bootstrap default
    # + fallback when the row is missing / malformed.
    CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT: int = 20

    @field_validator("CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT")
    @classmethod
    def _validate_sweep_drop_tolerance(cls, v: int) -> int:
        if not 0 <= v <= 100:
            raise ValueError(
                "CMDB_SYNC_SWEEP_DROP_TOLERANCE_PERCENT must be in [0, 100]"
            )
        return v

    # ── Scheduled-job execution concurrency ───────────────────────────
    # Max scheduled-job HANDLERS allowed to run concurrently in ONE
    # process (worker or API). Each running handler holds >=1 pooled DB
    # connection for its whole run, and some open a second nested session
    # for audit/notify — see ``database.py`` (pool_size=10 + max_overflow=20
    # = 30 connections per process).
    #
    # Why this exists (2026-07-19 incident): APScheduler's AsyncIOExecutor
    # fires every due cron as a bare ``loop.create_task`` with NO cross-job
    # cap (``max_instances=1`` is per-job only). When a cluster of nightly
    # cloud-sync crons (billing_daily + resource_sync across every cloud
    # account × capability) came due in the same minute, they checked out
    # all 30 connections at once; every other DB consumer — the sibling
    # crons AND their nested sessions — then failed to acquire a connection
    # within ``pool_timeout`` (30s) and died with
    # "QueuePool limit of size 10 overflow 20 reached, connection timed
    # out". The identical error across unrelated jobs, the AliCloud
    # ``domains`` capability timeout, and the identity-sync staleness
    # watchdog firing were all one event: pool exhaustion, not many bugs.
    #
    # Bounding concurrent executions well under the pool keeps headroom for
    # each handler's nested session + the worker's own housekeeping
    # sessions (reload loop, liveness watchdog). Default 8 → at most ~16
    # connections held by jobs, leaving ~14 of 30 free. Clamped to
    # ``[1, 25]`` so it can never be set at/above the pool ceiling.
    SCHEDULED_JOB_MAX_CONCURRENCY: int = 8

    @field_validator("SCHEDULED_JOB_MAX_CONCURRENCY")
    @classmethod
    def _validate_sched_job_concurrency(cls, v: int) -> int:
        if not 1 <= v <= 25:
            raise ValueError(
                "SCHEDULED_JOB_MAX_CONCURRENCY must be in [1, 25] — it must "
                "stay below the DB pool ceiling (pool_size 10 + overflow 20 "
                "= 30) with headroom for nested + housekeeping sessions."
            )
        return v

    # Sub-limit for ``weight="heavy"`` handlers (multi-minute cloud vendor
    # sweeps: cmdb.resource_sync / billing / scopes / roles / grants /
    # cost_facts, hris.full_sync). Heavy handlers acquire this gate FIRST
    # and then the global gate, so slow cloud syncs can occupy at most
    # this many of the global slots — fast critical crons (provisioning
    # dispatch, identity_sync dispatch, outbox retry) always have
    # ``MAX_CONCURRENCY - HEAVY`` slots they can never be starved out of.
    # Effective value is min(this, SCHEDULED_JOB_MAX_CONCURRENCY).
    SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY: int = 4

    @field_validator("SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY")
    @classmethod
    def _validate_sched_job_heavy_concurrency(cls, v: int) -> int:
        if not 1 <= v <= 25:
            raise ValueError(
                "SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY must be in [1, 25]; "
                "it is additionally clamped to SCHEDULED_JOB_MAX_CONCURRENCY "
                "at use time."
            )
        return v

    # Failure-storm aggregation (2026-07-19 post-mortem): when at least
    # this many DISTINCT scheduled jobs have a failed run inside the
    # rolling 10-minute window, the failure notifier collapses per-job
    # cards into ONE "systemic incident" card (P0-tier jobs still page
    # individually). Prevents a single infrastructure event (pool
    # exhaustion, Redis outage, TiDB brown-out) from fanning N near-
    # identical cards to every admin.
    SCHEDULED_JOB_FAILURE_STORM_THRESHOLD: int = 4

    @field_validator("SCHEDULED_JOB_FAILURE_STORM_THRESHOLD")
    @classmethod
    def _validate_sched_job_storm_threshold(cls, v: int) -> int:
        if not 2 <= v <= 50:
            raise ValueError(
                "SCHEDULED_JOB_FAILURE_STORM_THRESHOLD must be in [2, 50]."
            )
        return v

    # Minimum quiet period between two failure cards for the SAME job
    # (2026-08-17): a job whose error text flaps between two shapes
    # (``INTELLIGENCE_OUTPUT_INVALID`` / ``BAILIAN_LLM_BAD_JSON`` on the
    # daily AI brief) defeats signature dedup on every alternation and
    # pushed 3 P1 cards inside 36 minutes. Non-P0 tiers get at most one
    # card per this many minutes; an acknowledged prior failure still
    # re-opens the gate immediately, and P0 (paging tier) is exempt.
    SCHEDULED_JOB_FAILURE_NOTIFY_COOLDOWN_MINUTES: int = 60

    @field_validator("SCHEDULED_JOB_FAILURE_NOTIFY_COOLDOWN_MINUTES")
    @classmethod
    def _validate_sched_job_notify_cooldown(cls, v: int) -> int:
        if not 0 <= v <= 1440:
            raise ValueError(
                "SCHEDULED_JOB_FAILURE_NOTIFY_COOLDOWN_MINUTES must be in [0, 1440]."
            )
        return v

    # CMDB credential encryption
    CREDENTIAL_ENCRYPTION_KEY: str = ""  # base64url 32 bytes; required when managing provider credentials

    @field_validator("CREDENTIAL_ENCRYPTION_KEY")
    @classmethod
    def validate_encryption_key(cls, v: str) -> str:
        if not v:
            return v  # empty is allowed — cloud features will raise on use
        try:
            decoded = base64.urlsafe_b64decode(v + "==")  # pad for urlsafe_b64decode
            if len(decoded) < 32:
                raise ValueError("CREDENTIAL_ENCRYPTION_KEY must decode to at least 32 bytes")
        except Exception as exc:
            raise ValueError(f"CREDENTIAL_ENCRYPTION_KEY is not valid base64url: {exc}") from exc
        return v

    @field_validator("CORS_ORIGINS", mode="before")
    @classmethod
    def parse_cors_origins(cls, v: object) -> list[str]:
        if isinstance(v, str):
            try:
                parsed = json.loads(v)
                if isinstance(parsed, list):
                    return parsed
            except (json.JSONDecodeError, TypeError):
                pass
            return [s.strip() for s in v.split(",") if s.strip()]
        return v  # type: ignore[return-value]

    @field_validator("CORS_ORIGINS")
    @classmethod
    def reject_wildcard_cors(cls, v: list[str]) -> list[str]:
        """Pre-launch hardening 2026-05-04: refuse to start with
        ``CORS_ORIGINS=*`` outside of development. The runtime CRITICAL
        log was good observability but didn't prevent the process from
        booting and serving CORS=* with credentials=true (which the
        spec forbids). Now: typo in .env → process refuses to start.
        """
        import os

        if "*" not in v:
            return v
        env = (os.environ.get("APP_ENV") or "").lower()
        if env == "development":
            return v
        raise ValueError(
            "CORS_ORIGINS contains '*' — disallowed outside APP_ENV="
            "development. Tighten to your real frontend origin(s)."
        )

    # ── Prometheus / observability ─────────────────────────────────
    # Shared-secret gate for the public /metrics endpoint. When empty
    # the endpoint is OPEN (suitable for VPC-only / Docker-bridge
    # deployments where the metrics port is not internet-exposed).
    # When set, scrapers must send ``Authorization: Bearer <token>``
    # or the request returns 401. Match the prod ALB layout: the
    # endpoint is on the same FastAPI port as user APIs but lives
    # under /metrics, so a public ALB without VPC isolation MUST set
    # this. PR-C 2026-05-31.
    PLATFORM_METRICS_TOKEN: str = ""

    model_config = {
        "env_file": _default_env_files(),
        "env_file_encoding": "utf-8",
        "extra": "ignore",
    }


settings = Settings()