state-hub/api/services/legacy_compat.py

"""Shared legacy REST compat metering (routes and query params)."""
from __future__ import annotations

import logging

from fastapi import HTTPException, Request, Response, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.services.legacy_meter import identity_from_request, record_legacy_usage

logger = logging.getLogger(__name__)

_LEGACY_OWNER = "state-hub.api"
# Planning horizon for legacy workstream interfaces; actual removal remains
# legacy-meter-gated per docs/workplan-terminology-transition.md.
_LEGACY_WORKSTREAM_SUNSET = "Wed, 30 Jun 2027 23:59:59 GMT"


def legacy_response_headers(replacement_ref: str) -> dict[str, str]:
    return {
        "Deprecation": "true",
        "Sunset": _LEGACY_WORKSTREAM_SUNSET,
        "X-StateHub-Replacement": replacement_ref,
        "Link": f'<{replacement_ref}>; rel="successor-version"',
    }


def mark_legacy_response(response: Response | None, replacement_ref: str) -> None:
    if response is None:
        return
    for key, value in legacy_response_headers(replacement_ref).items():
        if key == "Link":
            response.headers.append(key, value)
        else:
            response.headers[key] = value


def legacy_query_param_key(method: str, route: str, param: str = "workstream_id") -> str:
    return f"rest_api:{method} {route}?{param}"


async def meter_legacy_query_param(
    *,
    session: AsyncSession,
    request: Request | None,
    response: Response | None,
    method: str,
    route: str,
    replacement_ref: str,
    param: str = "workstream_id",
) -> None:
    interface_key = legacy_query_param_key(method, route, param)
    mark_legacy_response(response, replacement_ref)
    try:
        await record_legacy_usage(
            session,
            interface_key=interface_key,
            interface_kind="rest_api",
            replacement_ref=replacement_ref,
            owner_component=_LEGACY_OWNER,
            replacement_verified=True,
            identity=identity_from_request(request),
        )
    except Exception:
        await session.rollback()
        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)


def legacy_body_field_key(method: str, route: str, field: str = "workstream_id") -> str:
    return f"rest_api:{method} {route} {field}"


async def meter_legacy_body_field(
    *,
    session: AsyncSession,
    request: Request | None,
    response: Response | None,
    method: str,
    route: str,
    replacement_ref: str,
    field: str = "workstream_id",
) -> None:
    interface_key = legacy_body_field_key(method, route, field)
    mark_legacy_response(response, replacement_ref)
    try:
        await record_legacy_usage(
            session,
            interface_key=interface_key,
            interface_kind="rest_api",
            replacement_ref=replacement_ref,
            owner_component=_LEGACY_OWNER,
            replacement_verified=True,
            identity=identity_from_request(request),
        )
    except Exception:
        await session.rollback()
        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)


async def retire_legacy_route(
    *,
    session: AsyncSession,
    request: Request | None,
    response: Response | None,
    interface_key: str,
    replacement_ref: str,
    detail: str,
) -> None:
    """Record final legacy-meter usage for this route, then return 410 Gone."""
    mark_legacy_response(response, replacement_ref)
    try:
        await record_legacy_usage(
            session,
            interface_key=interface_key,
            interface_kind="rest_api",
            replacement_ref=replacement_ref,
            owner_component=_LEGACY_OWNER,
            replacement_verified=True,
            identity=identity_from_request(request),
        )
    except Exception:
        await session.rollback()
        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
    raise HTTPException(
        status_code=status.HTTP_410_GONE,
        detail=detail,
        headers=legacy_response_headers(replacement_ref),
    )


async def meter_legacy_body_from_model(
    body: object,
    *,
    session: AsyncSession,
    request: Request | None,
    response: Response | None,
    method: str,
    route: str,
    replacement_ref: str,
    field: str = "workstream_id",
) -> None:
    if not getattr(body, "used_legacy_workstream_id", False):
        return
    await meter_legacy_body_field(
        session=session,
        request=request,
        response=response,
        method=method,
        route=route,
        replacement_ref=replacement_ref,
        field=field,
    )