han-app/VM2_services/codebase/services/bitrix-sync/app/main.py

from __future__ import annotations

import hmac
from contextlib import asynccontextmanager
from typing import Annotated

import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse
from sqlalchemy import text

from app.config import Settings, load_settings
from app.repository import Repository
from app.security import WebhookValidationError, parse_bounded_form, validate_webhook
from app.telemetry import (
    init_telemetry,
    instrument_fastapi,
    log_event,
    record_webhook,
    shutdown_telemetry,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_telemetry("bitrix-sync-api")
    app.state.repository = None
    try:
        settings = load_settings()
        app.state.settings = settings
        app.state.repository = (
            Repository(settings.database_url.get_secret_value(), settings.db_pool_size)
            if settings.enabled and settings.database_url
            else None
        )
        log_event(
            "service.started",
            "Bitrix sync API started",
            attributes={"mode": settings.mode},
        )
        yield
    finally:
        if app.state.repository:
            await app.state.repository.close()
        log_event("service.stopped", "Bitrix sync API stopped")
        shutdown_telemetry()


app = FastAPI(
    title="HAN Bitrix Sync",
    version="0.1.0",
    docs_url=None,
    redoc_url=None,
    lifespan=lifespan,
)
instrument_fastapi(app)


def settings(request: Request) -> Settings:
    return request.app.state.settings


def repository(request: Request) -> Repository:
    repo = request.app.state.repository
    if repo is None:
        raise HTTPException(status_code=503, detail="sync_disabled")
    return repo


async def require_service_token(
    request: Request,
    authorization: Annotated[str | None, Header()] = None,
) -> None:
    configured = settings(request).service_token
    expected = f"Bearer {configured.get_secret_value()}" if configured else ""
    if not authorization or not hmac.compare_digest(authorization, expected):
        raise HTTPException(status_code=401, detail="unauthorized")


@app.get("/health/live", include_in_schema=True)
async def live() -> dict[str, str]:
    return {"status": "live"}


@app.get("/health/ready", include_in_schema=True)
async def ready(request: Request) -> Response:
    config = settings(request)
    if not config.enabled:
        return JSONResponse(
            status_code=503,
            content={"status": "not_ready", "reason": "sync_disabled"},
        )
    repo = repository(request)
    if not await repo.ping():
        return JSONResponse(status_code=503, content={"status": "not_ready", "reason": "database"})
    return JSONResponse({"status": "ready", "mode": config.mode})


@app.get(
    "/internal/sync/v1/status",
    dependencies=[Depends(require_service_token)],
    include_in_schema=True,
)
async def sync_status(request: Request) -> dict:
    result = await repository(request).status()
    result["mode"] = settings(request).mode
    return result


@app.post("/bitrix/sync/webhook/contact", status_code=202, include_in_schema=True)
async def contact_webhook(
    request: Request,
    token: Annotated[str | None, Query(max_length=256)] = None,
    ID: Annotated[str | None, Query(pattern=r"^[1-9][0-9]{0,19}$")] = None,  # noqa: N803
) -> Response:
    return await _receive(request, "contact", {"token": token or "", "ID": ID or ""})


@app.post("/bitrix/sync/webhook/alert", status_code=202, include_in_schema=True)
async def alert_webhook(
    request: Request,
    token: Annotated[str | None, Query(max_length=256)] = None,
    ID: Annotated[str | None, Query(pattern=r"^[1-9][0-9]{0,19}$")] = None,  # noqa: N803
) -> Response:
    return await _receive(request, "alert", {"token": token or "", "ID": ID or ""})


async def _receive(request: Request, receiver: str, query: dict[str, str]) -> Response:
    try:
        config = settings(request)
        if not config.enabled:
            raise HTTPException(status_code=503, detail="sync_disabled")
        if request.headers.get("content-type", "").split(";", 1)[0].lower() != (
            "application/x-www-form-urlencoded"
        ):
            raise HTTPException(status_code=400, detail="invalid_content_type")
        content_length = request.headers.get("content-length")
        if content_length and (
            not content_length.isdigit() or int(content_length) > config.webhook_max_body_bytes
        ):
            raise HTTPException(status_code=413, detail="body_too_large")
        body = await request.body()
        if len(body) > config.webhook_max_body_bytes:
            raise HTTPException(status_code=413, detail="body_too_large")
        form = parse_bounded_form(body, max_fields=config.webhook_max_fields)
        # nginx overwrites X-Real-IP from the TCP peer after its CIDR check.
        source_ip = request.headers.get("x-real-ip") or (
            request.client.host if request.client else ""
        )
        alert_entity_type_id = (
            await _alert_entity_type(repository(request)) if receiver == "alert" else None
        )
        event = validate_webhook(
            receiver,
            query,
            form,
            source_ip,
            config,
            alert_entity_type_id=alert_entity_type_id,
        )
    except PermissionError as exc:
        record_webhook(receiver, "rejected")
        raise HTTPException(status_code=403, detail="forbidden") from exc
    except WebhookValidationError as exc:
        record_webhook(receiver, "rejected")
        raise HTTPException(status_code=400, detail="malformed_webhook") from exc
    except HTTPException:
        record_webhook(receiver, "rejected")
        raise
    except Exception:
        record_webhook(receiver, "error")
        raise
    await repository(request).insert_webhook(
        event.receiver_type, event.event_type, event.entity_id, event.source_ip
    )
    record_webhook(receiver, "accepted")
    return Response(status_code=202)


async def _alert_entity_type(repo: Repository) -> int | None:
    async with repo.engine.connect() as connection:
        value = (
            await connection.execute(
                text(
                    """
                    SELECT (value_json->>'entity_type_id')::integer
                    FROM bitrix_sync.settings
                    WHERE key='business_alerts' AND active=true AND validation_status='valid'
                    """
                )
            )
        ).scalar_one_or_none()
        return value


def run() -> None:
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",  # noqa: S104 - container-only port, not host-published
        port=8080,
        proxy_headers=False,
        access_log=False,
    )