from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    read_strategy: str = "A"
    database_url: str
    redis_url: str = "redis://redis:6379/0"
    db_pool_min: int = 10
    db_pool_max: int = 10
    cache_ttl_seconds: int = 30

settings = Settings()

import asyncpg
from .config import settings

pool: asyncpg.Pool | None = None

async def init_pool() -> None:
    global pool
    pool = await asyncpg.create_pool(
        dsn=settings.database_url,
        min_size=settings.db_pool_min,
        max_size=settings.db_pool_max,
        max_inactive_connection_lifetime=0,
        command_timeout=5,
    )

async def close_pool() -> None:
    if pool:
        await pool.close()

import orjson
from ..db import pool

SQL = """
SELECT
    s.numero, s.cliente_id, s.estado, s.causa,
    s.monto_estimado, s.monto_aprobado,
    s.fecha_ocurrencia, s.fecha_aviso,
    jsonb_build_object(
        'numero', p.numero,
        'producto', p.producto,
        'suma_asegurada', p.suma_asegurada
    ) AS poliza,
    COALESCE((
        SELECT jsonb_agg(jsonb_build_object(
            'tipo', h.tipo, 'descripcion', h.descripcion,
            'actor', h.actor, 'ocurrido_en', h.ocurrido_en
        ) ORDER BY h.ocurrido_en)
        FROM siniestros_w.hito h WHERE h.siniestro_id = s.id
    ), '[]'::jsonb) AS hitos,
    COALESCE((
        SELECT jsonb_agg(jsonb_build_object(
            'tipo', d.tipo, 'estado', d.estado, 'cargado_en', d.cargado_en
        ) ORDER BY d.cargado_en)
        FROM siniestros_w.documento d WHERE d.siniestro_id = s.id
    ), '[]'::jsonb) AS documentos,
    (
        SELECT jsonb_build_object(
            'perito', pe.perito, 'estado', pe.estado,
            'resultado', pe.resultado, 'agendado_para', pe.agendado_para
        )
        FROM siniestros_w.peritaje pe
        WHERE pe.siniestro_id = s.id
        ORDER BY pe.id DESC LIMIT 1
    ) AS peritaje
FROM siniestros_w.siniestro s
JOIN siniestros_w.poliza p ON p.id = s.poliza_id
WHERE s.id = $1
"""

async def get_estado(siniestro_id: int) -> dict | None:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(SQL, siniestro_id)
    if row is None:
        return None
    data = dict(row)
    for k in ("poliza", "hitos", "documentos", "peritaje"):
        if isinstance(data[k], str):
            data[k] = orjson.loads(data[k])
    return data

import orjson
from ..db import pool

SQL = "SELECT payload FROM siniestros_r.siniestro_estado WHERE siniestro_id = $1"

async def get_estado(siniestro_id: int) -> dict | None:
    async with pool.acquire() as conn:
        raw = await conn.fetchval(SQL, siniestro_id)
    return orjson.loads(raw) if raw else None

import orjson
from ..cache import redis_client
from ..config import settings
from ..metrics import CACHE_HITS, CACHE_MISSES
from . import arm_b

def key(siniestro_id: int) -> str:
    return f"siniestro:estado:{siniestro_id}"

async def get_estado(siniestro_id: int) -> dict | None:
    cached = await redis_client.get(key(siniestro_id))
    if cached is not None:
        CACHE_HITS.inc()
        return orjson.loads(cached)

    CACHE_MISSES.inc()
    data = await arm_b.get_estado(siniestro_id)
    if data is not None:
        await redis_client.set(
            key(siniestro_id), orjson.dumps(data),
            ex=settings.cache_ttl_seconds,
        )
    return data


async def get_estado_raw(siniestro_id: int) -> bytes | None:
    """Brazo C': devuelve bytes JSON sin re-serializar (bypass de Pydantic)."""
    cached = await redis_client.get(key(siniestro_id))
    if cached is not None:
        CACHE_HITS.inc()
        return cached
    CACHE_MISSES.inc()
    data = await arm_b.get_estado(siniestro_id)
    if data is None:
        return None
    raw = orjson.dumps(data)
    await redis_client.set(key(siniestro_id), raw, ex=settings.cache_ttl_seconds)
    return raw

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import ORJSONResponse
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest

from .config import settings
from .db import init_pool, close_pool
from .cache import init_cache, close_cache
from .metrics import REQUEST_LATENCY
from .models import EstadoSiniestro
from .strategies import arm_a, arm_b, arm_c

STRATEGIES = {"A": arm_a.get_estado, "B": arm_b.get_estado, "C": arm_c.get_estado}

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_pool()
    await init_cache()
    yield
    await close_pool()
    await close_cache()

app = FastAPI(lifespan=lifespan, default_response_class=ORJSONResponse)

@app.get("/health")
async def health():
    return {"status": "ok", "arm": settings.read_strategy}

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.get("/siniestros/{siniestro_id}/estado", response_model=EstadoSiniestro)
async def estado_siniestro(siniestro_id: int):
    arm = settings.read_strategy
    with REQUEST_LATENCY.labels(arm=arm).time():
        if arm == "C_PRIME":
            raw = await arm_c.get_estado_raw(siniestro_id)
            if raw is None:
                raise HTTPException(404, "Siniestro no encontrado")
            return Response(content=raw, media_type="application/json")

        data = await STRATEGIES[arm](siniestro_id)
        if data is None:
            raise HTTPException(404, "Siniestro no encontrado")
        return data

from prometheus_client import Counter, Histogram

REQUEST_LATENCY = Histogram(
    "ha08_read_latency_seconds",
    "Latencia interna de la consulta de estado de siniestro",
    labelnames=("arm",),
    buckets=(0.005, 0.010, 0.025, 0.050, 0.075, 0.100,
             0.150, 0.250, 0.500, 1.0, 2.5),
)
CACHE_HITS = Counter("ha08_cache_hits_total", "Aciertos de caché")
CACHE_MISSES = Counter("ha08_cache_misses_total", "Fallos de caché")

import asyncio, os, orjson
import asyncpg
from aiokafka import AIOKafkaConsumer
from redis.asyncio import Redis
from prometheus_client import Histogram, Counter, start_http_server

LAG = Histogram(
    "ha08_projection_lag_seconds",
    "Lag entre ocurrencia del evento y proyección",
    buckets=(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30),
)
PROJECTED = Counter("ha08_events_projected_total", "Eventos proyectados")
SKIPPED = Counter("ha08_events_skipped_total", "Eventos descartados por versión")

UPSERT = """
INSERT INTO siniestros_r.siniestro_estado
       (siniestro_id, cliente_id, version, ocurrido_en, proyectado_en, payload)
VALUES ($1, $2, $3, $4, now(), $5)
ON CONFLICT (siniestro_id) DO UPDATE
SET cliente_id    = EXCLUDED.cliente_id,
    version       = EXCLUDED.version,
    ocurrido_en   = EXCLUDED.ocurrido_en,
    proyectado_en = now(),
    payload       = EXCLUDED.payload
WHERE siniestros_r.siniestro_estado.version < EXCLUDED.version
RETURNING siniestro_id
"""

async def main() -> None:
    start_http_server(8001)
    pool = await asyncpg.create_pool(os.environ["DATABASE_URL"], min_size=2, max_size=5)
    redis = Redis.from_url(os.environ["REDIS_URL"])
    consumer = AIOKafkaConsumer(
        os.environ["TOPIC"],
        bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"],
        group_id=os.environ["CONSUMER_GROUP"],
        enable_auto_commit=True,
        auto_offset_reset="latest",
    )
    await consumer.start()
    try:
        async for msg in consumer:
            ev = orjson.loads(msg.value)
            async with pool.acquire() as conn:
                applied = await conn.fetchval(
                    UPSERT,
                    ev["siniestro_id"], ev["payload"]["cliente_id"],
                    ev["version"], ev["ocurrido_en"],
                    orjson.dumps(ev["payload"]).decode(),
                )
            if applied is None:
                SKIPPED.inc()
                continue
            # Invalidación proactiva: el caché no debe servir estado vencido
            await redis.delete(f"siniestro:estado:{ev['siniestro_id']}")
            PROJECTED.inc()
            LAG.observe(_lag_seconds(ev["ocurrido_en"]))
    finally:
        await consumer.stop()
        await pool.close()
        await redis.aclose()

if __name__ == "__main__":
    asyncio.run(main())

# 1. elegir un siniestro del hot set (los que un cliente consultaría)
# 2. UPDATE siniestros_w.siniestro SET estado=..., version = version + 1
# 3. INSERT en siniestros_w.hito
# 4. construir el payload completo (mismo shape que arm_a)
# 5. publicar en Redpanda con key=str(siniestro_id) y ocurrido_en=now()