Errores, logging y tareas idempotentes

"""Códigos de error como constantes con nombre (igual que shared/errors/codes.py de m-b-core)."""

# 400 — validation
VALIDATION_ERROR = "VALIDATION_ERROR"
# 401
INVALID_CRON_SECRET = "INVALID_CRON_SECRET"
# 404
LOCAL_NOT_FOUND = "LOCAL_NOT_FOUND"
# 409 — business rule
STOCK_INSUFICIENTE = "STOCK_INSUFICIENTE"
# 503
UPSTREAM_UNAVAILABLE = "UPSTREAM_UNAVAILABLE"
# 500
INTERNAL_ERROR = "INTERNAL_ERROR"

"""Excepciones de dominio.

Se lanzan desde services/queries; NUNCA saben de HTTP. La traducción a status +
error_response ocurre una sola vez, en los exception handlers de main.py.
"""

from __future__ import annotations

from . import codes


class DomainError(Exception):
    """Base: toda excepción de negocio tiene código y status por defecto."""

    code: str = codes.INTERNAL_ERROR
    status_code: int = 500

    def __init__(self, message: str, *, detail: str | None = None) -> None:
        super().__init__(message)
        self.message = message
        self.detail = detail


class LocalNoExiste(DomainError):
    code = codes.LOCAL_NOT_FOUND
    status_code = 404


class StockInsuficiente(DomainError):
    code = codes.STOCK_INSUFICIENTE
    status_code = 409


class UpstreamNoDisponible(DomainError):
    code = codes.UPSTREAM_UNAVAILABLE
    status_code = 503

"""error_response(): misma forma que m-b-core (shared/errors/responses.py)."""

from __future__ import annotations

from enum import Enum
from typing import Any


def error_response(
    error_code: str | Enum,
    message: str,
    status_code: int = 400,
    detail: str | None = None,
) -> dict[str, Any]:
    error_value = error_code.value if isinstance(error_code, Enum) else str(error_code)
    response: dict[str, Any] = {"success": False, "error": error_value, "message": message}
    if detail:
        response["detail"] = detail
    return response

from .exceptions import DomainError, LocalNoExiste, StockInsuficiente, UpstreamNoDisponible
from .responses import error_response

__all__ = [
    "DomainError",
    "LocalNoExiste",
    "StockInsuficiente",
    "UpstreamNoDisponible",
    "error_response",
]

"""API: app + lifespan + exception handlers (una sola traducción error → HTTP)."""

from __future__ import annotations

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from shared.errors import DomainError, codes, error_response
from shared.logging_setup import configure_logging
from shared.settings import settings

from .middleware.request_id import request_id_middleware
from .router import build_api_router

configure_logging(settings.LOG_LEVEL)
log = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(settings.UPSTREAM_TIMEOUT_S))
    log.info("api.start", extra={"version": settings.APP_VERSION, "env": settings.ENV})
    yield
    await app.state.http.aclose()


app = FastAPI(
    title="mesa-core-practicas",
    lifespan=lifespan,
    docs_url=None if settings.ENV == "prod" else "/docs",
)
app.middleware("http")(request_id_middleware)
app.include_router(build_api_router(), prefix="/v1")


@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError) -> JSONResponse:
    # 4xx de negocio: WARNING (esperable). 5xx de dominio (upstream caído): ERROR con la causa
    # resumida (el `from e`); el traceback completo se reserva para lo NO manejado (bugs).
    level = logging.ERROR if exc.status_code >= 500 else logging.WARNING
    log.log(
        level,
        "domain.error",
        extra={"code": exc.code, "path": request.url.path, "cause": repr(exc.__cause__)},
    )
    return JSONResponse(
        status_code=exc.status_code,
        content=error_response(exc.code, exc.message, exc.status_code, exc.detail),
    )


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        status_code=422,
        content=error_response(
            codes.VALIDATION_ERROR, "Entrada inválida", 422, str(exc.errors()[0]["msg"])
        ),
    )


@app.exception_handler(Exception)
async def unhandled_handler(request: Request, exc: Exception) -> JSONResponse:
    # Lo que no es de dominio es un bug: ERROR con traza al log, mensaje genérico al cliente.
    log.error("unhandled", extra={"path": request.url.path}, exc_info=exc)
    return JSONResponse(
        status_code=500, content=error_response(codes.INTERNAL_ERROR, "Error interno", 500)
    )


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok", "version": settings.APP_VERSION}

"""Un id por request: llega en X-Request-ID (el LB lo pone) o se genera; se propaga a los logs."""

from __future__ import annotations

import uuid
from collections.abc import Awaitable, Callable

from fastapi import Request, Response

from shared.logging_setup import request_id_var


async def request_id_middleware(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:12]
    token = request_id_var.set(rid)
    try:
        response = await call_next(request)
    finally:
        request_id_var.reset(token)
    response.headers["X-Request-ID"] = rid
    return response

@router.get("/locales/{local_id}", response_model=LocalDTO)
async def get_local(local_id: int) -> LocalDTO:
    return LocalDTO(**await queries.fetch_local(local_id))  # LocalNoExiste → 404 en el handler


@router.post("/locales/{local_id}/asignar", response_model=ReservaOut, status_code=201)
async def post_asignar(
    local_id: int, request: Request, pax: Annotated[int, Query(ge=1, le=20)]
) -> ReservaOut:
    result = await asignar_mesa(local_id, pax, _client(request))
    return ReservaOut(reserva_id=1, **result)

async def fetch_local(local_id: int) -> dict[str, Any]:
    """Devuelve el local o lanza LocalNoExiste. Nunca devuelve None 'para que el caller vea'."""
    row = _LOCALES.get(local_id)
    if row is None:
        raise LocalNoExiste(f"Local {local_id} no existe")
    return row


async def fetch_capacidad_upstream(local_id: int, client: httpx.AsyncClient) -> int:
    """Consulta un servicio externo. Un fallo de red se TRADUCE a excepción de dominio."""
    try:
        r = await client.get(f"{settings.UPSTREAM_URL}/capacidad/{local_id}")
        r.raise_for_status()
    except httpx.HTTPError as e:  # concreta: timeout, conexión, 5xx del upstream
        raise UpstreamNoDisponible("servicio de capacidad no disponible") from e
    return int(r.json()["capacidad"])

[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "C4", "UP", "BLE", "G", "DTZ", "S608"]   # ← BLE y G nuevos

"""Ejemplo a propósito de lo que ruff debe atrapar. Se borra después."""

import logging

import httpx

log = logging.getLogger(__name__)


async def capacidad(local_id: int, client: httpx.AsyncClient) -> int:
    try:
        r = await client.get(f"http://upstream/capacidad/{local_id}")
    except Exception as e:  # vicio 1: traga todo y sigue
        print(f"error: {e}")  # vicio 3: print
        return 0
    log.info(f"capacidad ok local={local_id}")  # vicio: f-string en el log
    log.warning("upstream lento: %s" % r.status_code)  # vicio: % en el log
    return int(r.json()["capacidad"])

def test_503_upstream_caido_no_es_500_ni_200(client):
    from shared.errors import UpstreamNoDisponible

    with patch(_CAP, new=AsyncMock(side_effect=UpstreamNoDisponible("caído"))):
        r = client.post("/v1/locales/11/asignar?pax=2")
    assert r.status_code == 503
    assert r.json()["error"] == "UPSTREAM_UNAVAILABLE"


def test_bug_no_manejado_da_500_generico_sin_traza(caplog):
    # TestClient relanza las excepciones del servidor por defecto (para que el test las vea);
    # aquí queremos probar el handler de 500, así que se lo pedimos explícito.
    with (
        TestClient(app, raise_server_exceptions=False) as client,
        patch(_CAP, new=AsyncMock(side_effect=ZeroDivisionError("boom"))),
    ):
        r = client.post("/v1/locales/11/asignar?pax=2")
    assert r.status_code == 500
    assert r.json() == {"success": False, "error": "INTERNAL_ERROR", "message": "Error interno"}
    assert "boom" not in r.text  # el detalle va al log, no al cliente
    assert any(x.levelno == logging.ERROR and x.exc_info for x in caplog.records)

"""Logging estructurado: JSON por línea, campos en `extra`, PII redactada.

Guideline 3 de m-b-core/CLAUDE.md: INFO eventos de negocio, DEBUG detalle,
WARNING dato raro, ERROR se rompió; SIEMPRE `extra={}`; nunca interpolar.
"""

from __future__ import annotations

import json
import logging
import re
import sys
from contextvars import ContextVar
from datetime import UTC, datetime
from typing import Any

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

_STD_ATTRS = set(logging.LogRecord("x", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"}
# Un email de verdad termina en un TLD con letras (así "python@3.12_4/…" en una ruta no matchea).
_EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)*\.[A-Za-z]{2,}")
# Teléfono: solo con prefijo internacional (+51 987 654 321); los timestamps no lo tienen.
_PHONE = re.compile(r"\+\d{1,3}(?:[\s-]?\d{2,4}){2,4}")
_BEARER = re.compile(r"(?i)bearer\s+[A-Za-z0-9._-]+")
# Campos que SIEMPRE son PII, se llamen como se llamen sus valores.
_PII_KEYS = {"email", "correo", "telefono", "phone", "celular", "destinatario", "to"}
_SKIP_KEYS = {"ts", "level", "logger", "request_id"}


def redact(value: Any, key: str | None = None) -> Any:
    """Redacta PII: por nombre de campo y por patrón; recursivo en dict/list."""
    if key in _PII_KEYS and value is not None:
        return "<redacted>"
    if isinstance(value, str):
        value = _EMAIL.sub("<email>", value)
        value = _BEARER.sub("Bearer <redacted>", value)
        return _PHONE.sub("<phone>", value)
    if isinstance(value, dict):
        return {k: (v if k in _SKIP_KEYS else redact(v, k)) for k, v in value.items()}
    if isinstance(value, list):
        return [redact(v) for v in value]
    return value


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": datetime.now(UTC).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "event": record.getMessage(),
            "request_id": request_id_var.get(),
        }
        for key, val in record.__dict__.items():  # los campos de extra={}
            if key not in _STD_ATTRS and not key.startswith("_"):
                payload[key] = val
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(redact(payload), ensure_ascii=False, default=str)


def configure_logging(level: str = "INFO") -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(level)

import json
import logging

from shared.logging_setup import JsonFormatter, redact, request_id_var


def test_redact_email_telefono_bearer():
    assert redact("ana.perez@gmail.com llamó desde +51 987 654 321 con Bearer eyJabc.def") == (
        "<email> llamó desde <phone> con Bearer <redacted>"
    )


def test_no_redacta_timestamps_ni_rutas_con_arroba():
    assert redact("2026-08-17T12:07:49.277+00:00") == "2026-08-17T12:07:49.277+00:00"
    assert redact("/opt/homebrew/Cellar/python@3.12/bin") == "/opt/homebrew/Cellar/python@3.12/bin"


def test_redacta_por_nombre_de_campo():
    assert redact({"destinatario": "cualquier cosa", "pax": 2}) == {
        "destinatario": "<redacted>",
        "pax": 2,
    }


def test_formatter_emite_json_con_extra_y_request_id():
    request_id_var.set("req-1")
    rec = logging.LogRecord("t", logging.INFO, "", 0, "reserva.creada", (), None)
    rec.local_id = 11
    rec.email = "ana@example.com"
    out = json.loads(JsonFormatter().format(rec))
    assert out["event"] == "reserva.creada" and out["request_id"] == "req-1"
    assert out["local_id"] == 11
    assert out["email"] == "<redacted>"  # por nombre de campo, antes que por patrón

def test_201_feliz_y_log_con_campos(client, caplog):
    caplog.set_level(logging.INFO)
    with patch(_CAP, new=AsyncMock(return_value=8)):
        r = client.post("/v1/locales/11/asignar?pax=2", headers={"X-Request-ID": "req-abc"})
    assert r.status_code == 201
    rec = next(x for x in caplog.records if x.getMessage() == "asignacion.ok")
    assert rec.local_id == 11 and rec.pax == 2 and rec.cap == 8  # campos de extra, no texto

async def asignar_mesa(local_id: int, pax: int, client: httpx.AsyncClient) -> dict[str, int]:
    local = await queries.fetch_local(local_id)  # puede lanzar LocalNoExiste
    capacidad = await queries.fetch_capacidad_upstream(local_id, client)  # UpstreamNoDisponible
    if pax > capacidad:
        # WARNING: dato inesperado del negocio, no un bug. Campos en extra, no en el mensaje.
        log.warning(
            "asignacion.sin_capacidad", extra={"local_id": local_id, "pax": pax, "cap": capacidad}
        )
        raise StockInsuficiente(f"No hay mesa para {pax} personas", detail=f"capacidad={capacidad}")
    log.info("asignacion.ok", extra={"local_id": local["id"], "pax": pax, "cap": capacidad})
    return {"local_id": local_id, "pax": pax, "mesa_asignada": 1}

"""Configuración centralizada: el ÚNICO módulo que lee variables de entorno.

m-b-core hoy usa una clase plana que hace `os.getenv` en import (shared/settings.py).
Aquí usamos pydantic-settings, que ya está en sus requirements: valida tipos,
falla al arrancar si falta algo obligatorio y documenta los defaults.
"""

from __future__ import annotations

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    ENV: str = "dev"
    SERVICE_NAME: str = "api"
    APP_VERSION: str = "dev"
    LOG_LEVEL: str = "INFO"
    DB_URL: str = "sqlite+aiosqlite:///./practicas.db"
    UPSTREAM_URL: str = "http://127.0.0.1:9999"
    UPSTREAM_TIMEOUT_S: float = 5.0
    # Secreto compartido para tareas (Cloud Scheduler lo manda como X-Cron-Secret).
    # Sin default: si falta, el worker no arranca. Nunca `= ""` (postmortem parallevar).
    CRON_SECRET: str = Field(..., min_length=16)


settings = Settings()  # type: ignore[call-arg]

"""POST /tasks/recordatorios — lo invoca Cloud Scheduler con X-Cron-Secret.

Comparación en tiempo constante (hmac.compare_digest). m-b-core hoy compara con `!=`
(services/api/app/admin/_deps.py:19, worker/app/modules/legacy_replica/router.py:74).
"""

from __future__ import annotations

import hmac
import logging
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Header, HTTPException, Request, status

from shared.db import get_session
from shared.errors import codes, error_response
from shared.settings import settings

from .notificador import NotificadorConsola
from .service import Resultado, enviar_recordatorios

log = logging.getLogger(__name__)
router = APIRouter(prefix="/tasks", tags=["tasks"])

# En la vida real vienen de la BD legacy (legacy_db_read); aquí una lista.
RESERVAS_MANANA: list[dict[str, Any]] = [
    {"id": 501, "email": "ana@example.com", "hora": "20:00", "pax": 2},
    {"id": 502, "email": "luis@example.com", "hora": "21:00", "pax": 4},
]


def require_cron_secret(x_cron_secret: Annotated[str | None, Header()] = None) -> None:
    presented = x_cron_secret or ""
    if not hmac.compare_digest(presented.encode(), settings.CRON_SECRET.encode()):
        log.warning("tasks.auth_fail", extra={"header_present": bool(x_cron_secret)})
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=error_response(codes.INVALID_CRON_SECRET, "Cron secret inválido", 401),
        )


@router.post("/recordatorios", dependencies=[Depends(require_cron_secret)])
async def run_recordatorios(request: Request) -> dict[str, int]:
    notificador = getattr(request.app.state, "notificador", None) or NotificadorConsola()
    async with get_session() as session:
        res: Resultado = await enviar_recordatorios(session, RESERVAS_MANANA, notificador)
    return {"vistas": res.vistas, "enviadas": res.enviadas, "omitidas": res.omitidas}

def require_admin_key(x_admin_key: Annotated[str, Header()]) -> None:
    """Reject requests without a valid `X-Admin-Key` matching SEND_API_KEY."""
    if not settings.SEND_API_KEY:
        # Misconfiguration — admin surface effectively disabled. Rare, log loud.
        logger.error("admin.auth.no_key_configured")
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid admin key.")
    if x_admin_key != settings.SEND_API_KEY:      # ← comparación NO constante
        logger.warning("admin.auth.invalid_key")
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid admin key.")

class Aviso(SQLModel, table=True):
    """Un aviso enviado. La restricción única (reserva, tipo) ES la idempotencia."""

    __table_args__ = (UniqueConstraint("reserva_id", "tipo", name="uq_aviso_reserva_tipo"),)

    id: int | None = Field(default=None, primary_key=True)
    reserva_id: int = Field(index=True)
    tipo: str
    enviado_en: datetime = Field(default_factory=_utcnow)


class Latido(SQLModel, table=True):
    """Última corrida de cada tarea: quien vigila mira este reloj."""

    tarea: str = Field(primary_key=True)
    ultima_corrida: datetime = Field(default_factory=_utcnow)
    avisos_enviados: int = 0

@retry(
    retry=retry_if_exception_type(EnvioFallido),
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=0.1, max=2),
    reraise=True,
)
async def _enviar_con_reintentos(notificador: Notificador, destinatario: str, texto: str) -> None:
    try:
        await notificador.enviar(destinatario, texto)
    except Exception as e:  # noqa: BLE001 — el proveedor puede fallar de mil formas; las envolvemos
        log.warning("notificacion.reintento", extra={"error": type(e).__name__})
        raise EnvioFallido(str(e)) from e


async def enviar_recordatorios(
    session: AsyncSession, reservas: list[dict[str, Any]], notificador: Notificador
) -> Resultado:
    """1) ¿ya avisé? → omitir. 2) enviar (con reintentos). 3) registrar. Nunca al revés."""
    enviadas = omitidas = 0
    for r in reservas:
        ya = await session.execute(
            select(Aviso).where(Aviso.reserva_id == r["id"], Aviso.tipo == TIPO_RECORDATORIO)
        )
        if ya.scalar_one_or_none() is not None:
            omitidas += 1
            continue
        await _enviar_con_reintentos(
            notificador, r["email"], f"Te esperamos mañana a las {r['hora']} para {r['pax']}."
        )
        session.add(Aviso(reserva_id=r["id"], tipo=TIPO_RECORDATORIO))
        try:
            await session.commit()  # la UNIQUE cierra la carrera entre dos corridas simultáneas
        except IntegrityError:
            await session.rollback()
            omitidas += 1
            log.warning("aviso.duplicado_concurrente", extra={"reserva_id": r["id"]})
            continue
        enviadas += 1
    latido = await session.get(Latido, "recordatorios") or Latido(tarea="recordatorios")
    latido.ultima_corrida = datetime.now(UTC)
    latido.avisos_enviados = enviadas
    session.add(latido)
    await session.commit()
    log.info(
        "recordatorios.done",
        extra={"vistas": len(reservas), "enviadas": enviadas, "omitidas": omitidas},
    )
    return Resultado(vistas=len(reservas), enviadas=enviadas, omitidas=omitidas)

@pytest.fixture()
async def session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with maker() as s:
        yield s
    await engine.dispose()


async def test_tres_corridas_un_solo_envio(session):
    n = NotificadorConsola()
    r1 = await enviar_recordatorios(session, RESERVAS, n)
    r2 = await enviar_recordatorios(session, RESERVAS, n)
    r3 = await enviar_recordatorios(session, RESERVAS, n)
    assert (r1.enviadas, r2.enviadas, r3.enviadas) == (2, 0, 0)
    assert (r2.omitidas, r3.omitidas) == (2, 2)
    assert len(n.enviados) == 2
    avisos = (await session.execute(select(Aviso))).scalars().all()
    assert len(avisos) == 2


async def test_reintenta_y_luego_envia(session):
    class Flaky:
        def __init__(self, fallos: int) -> None:
            self.fallos, self.calls = fallos, 0
            self.enviados: list[str] = []

        async def enviar(self, destinatario: str, texto: str) -> None:
            self.calls += 1
            if self.calls <= self.fallos:
                raise ConnectionError("proveedor caído")
            self.enviados.append(destinatario)

    n = Flaky(fallos=2)
    r = await enviar_recordatorios(session, RESERVAS[:1], n)
    assert r.enviadas == 1 and n.calls == 3


async def test_reintentos_agotados_no_registra_aviso(session):
    class Siempre:
        async def enviar(self, destinatario: str, texto: str) -> None:
            raise ConnectionError("no")

    with pytest.raises(EnvioFallido):
        await enviar_recordatorios(session, RESERVAS[:1], Siempre())
    assert (await session.execute(select(Aviso))).scalars().all() == []

# Recordatorios 24 h

Cron: todos los días a las 10:00 hora de Lima.
Endpoint: POST /tasks/recordatorios  ·  Auth: X-Cron-Secret
Idempotente: sí (tabla Aviso, UNIQUE reserva_id+tipo). Reintentar es seguro.

## Job de Cloud Scheduler (transitorio: secreto compartido)

    gcloud scheduler jobs create http recordatorios-24h \
      --location=us-central1 \
      --schedule="0 10 * * *" --time-zone="America/Lima" \
      --uri="https://WORKER_URL/tasks/recordatorios" \
      --http-method=POST \
      --headers="X-Cron-Secret=$(gcloud secrets versions access latest --secret=CRON_SECRET)" \
      --attempt-deadline=300s \
      --max-retry-attempts=3 --min-backoff=30s --max-backoff=300s

## Lo que deberíamos usar (identidad, no secreto compartido)

    gcloud scheduler jobs create http recordatorios-24h \
      --location=us-central1 \
      --schedule="0 10 * * *" --time-zone="America/Lima" \
      --uri="https://WORKER_URL/tasks/recordatorios" --http-method=POST \
      --oidc-service-account-email=cron-recordatorios@PROJECT.iam.gserviceaccount.com \
      --oidc-token-audience="https://WORKER_URL"

El worker valida el token OIDC y la audiencia; no hay secreto que rotar ni
que aparezca en `gcloud scheduler jobs describe`. Con --headers, el secreto
es legible por cualquiera con roles/cloudscheduler.viewer.

## Cola de Cloud Tasks (si el fan-out crece)

    gcloud tasks queues create recordatorios --location=us-central1 \
      --max-attempts=5 --min-backoff=10s --max-backoff=300s \
      --max-dispatches-per-second=10