"""Closed, local-only metering and Showback contracts.

The module intentionally has no network providers, price book, billing action
or financial artifact.  It only normalizes engineering evidence and supplies
integer-micros arithmetic for the database gateway and local fixture.
"""

from __future__ import annotations

import copy
import hashlib
import json
import os
import re
import unicodedata
import uuid
from datetime import UTC, datetime
from decimal import ROUND_HALF_EVEN, Decimal, InvalidOperation
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.pool import NullPool

from app.config.database_urls import validate_postgresql_url

EVENT_KINDS = frozenset(
    {"query", "api", "file", "subscription", "storage", "compute", "task", "model_call"}
)
_EVENT_FIELDS = frozenset(
    {
        "schema_version",
        "event_uid",
        "event_kind",
        "occurred_at",
        "window_start",
        "window_end",
        "quantity",
        "unit",
        "idempotency_key",
        "evidence",
        "mapping",
        "correction_of",
    }
)
_TOKEN = re.compile(r"[a-z][a-z0-9-]{0,62}")
_UID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,119}")
_HEX = re.compile(r"[0-9a-f]{64}")
_DECIMAL = re.compile(r"(?:0|[1-9][0-9]{0,17})(?:\.[0-9]{1,6})?")
_REFERENCE = re.compile(r"local-fixture://wp12/v1(?:/[a-z][a-z0-9-]{0,62})?")
_UNSAFE_REFERENCE_SEGMENT = re.compile(r"(?:sql|script|select|insert|update|delete|drop|exec|curl|wget|token|secret|credential|password)")
_UNITS: dict[str, tuple[str, Decimal]] = {
    "bytes": ("data", Decimal("1")),
    "kb": ("data", Decimal("1000")),
    "mb": ("data", Decimal("1000000")),
    "gb": ("data", Decimal("1000000000")),
    "seconds": ("time", Decimal("1")),
    "milliseconds": ("time", Decimal("0.001")),
    "requests": ("count", Decimal("1")),
    "tasks": ("count", Decimal("1")),
    "tokens": ("count", Decimal("1")),
}
_MICRO = Decimal("1000000")


def _closed(value: Any, allowed: frozenset[str], label: str) -> dict[str, Any]:
    if not isinstance(value, dict) or set(value) - allowed:
        raise ValueError(f"{label}_closed")
    return copy.deepcopy(value)


def _normalized_text(value: Any, label: str, maximum: int = 120) -> str:
    if not isinstance(value, str) or not value or len(value) > maximum:
        raise ValueError(f"{label}_invalid")
    normalized = unicodedata.normalize("NFKC", value)
    if normalized != value:
        raise ValueError(f"{label}_unicode_rejected")
    return value


def _token(value: Any, label: str) -> str:
    result = _normalized_text(value, label)
    if not _TOKEN.fullmatch(result):
        raise ValueError(f"{label}_invalid")
    return result


def _uid(value: Any, label: str) -> str:
    result = _normalized_text(value, label)
    if not _UID.fullmatch(result):
        raise ValueError(f"{label}_invalid")
    return result


def _timestamp(value: Any, label: str) -> str:
    text = _normalized_text(value, label, 30)
    if not text.endswith("Z"):
        raise ValueError(f"{label}_invalid")
    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ValueError(f"{label}_invalid") from exc
    if parsed.tzinfo is None or parsed.utcoffset() != UTC.utcoffset(parsed):
        raise ValueError(f"{label}_invalid")
    return parsed.astimezone(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z")


def _decimal(value: Any, label: str) -> Decimal:
    if isinstance(value, bool) or not isinstance(value, str) or not _DECIMAL.fullmatch(value):
        raise ValueError(f"{label}_invalid")
    try:
        parsed = Decimal(value)
    except InvalidOperation as exc:
        raise ValueError(f"{label}_invalid") from exc
    if not parsed.is_finite() or parsed < 0:
        raise ValueError(f"{label}_invalid")
    return parsed


def to_micros(quantity: Any, source_unit: Any, target_unit: Any) -> int:
    """Convert a bounded string decimal using a compatible unit dimension."""

    value = _decimal(quantity, "quantity")
    source = _normalized_text(source_unit, "source_unit", 20)
    target = _normalized_text(target_unit, "target_unit", 20)
    if source not in _UNITS or target not in _UNITS or _UNITS[source][0] != _UNITS[target][0]:
        raise ValueError("unit_dimension_mismatch")
    converted = value * _UNITS[source][1] / _UNITS[target][1]
    micros = (converted * _MICRO).quantize(Decimal("1"), rounding=ROUND_HALF_EVEN)
    if micros < 0 or micros > Decimal("1000000000000000000000000"):
        raise ValueError("quantity_out_of_range")
    return int(micros)


def canonical_digest(value: dict[str, Any]) -> str:
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def normalize_event(value: Any) -> dict[str, Any]:
    """Return a versioned, safe event without raw source evidence."""

    body = _closed(value, _EVENT_FIELDS, "metering_event")
    required = _EVENT_FIELDS - {"correction_of"}
    if set(body) - {"correction_of"} != required:
        raise ValueError("metering_event_closed")
    if body["schema_version"] != 1:
        raise ValueError("schema_version_invalid")
    event_kind = _token(body["event_kind"], "event_kind")
    if event_kind not in EVENT_KINDS:
        raise ValueError("event_kind_invalid")
    evidence = _closed(body["evidence"], frozenset({"digest", "reference"}), "evidence")
    digest = _normalized_text(evidence.get("digest"), "evidence_digest", 64)
    reference = _normalized_text(evidence.get("reference"), "evidence_reference", 120)
    if not _HEX.fullmatch(digest) or not _REFERENCE.fullmatch(reference) or _UNSAFE_REFERENCE_SEGMENT.search(reference):
        raise ValueError("evidence_invalid")
    mapping = _closed(
        body["mapping"],
        frozenset({"department", "business_domain", "project", "cost_center"}),
        "mapping",
    )
    if set(mapping) != {"department", "business_domain", "project", "cost_center"}:
        raise ValueError("mapping_closed")
    normalized = {
        "schema_version": 1,
        "event_uid": _uid(body["event_uid"], "event_uid"),
        "event_kind": event_kind,
        "occurred_at": _timestamp(body["occurred_at"], "occurred_at"),
        "window_start": _timestamp(body["window_start"], "window_start"),
        "window_end": _timestamp(body["window_end"], "window_end"),
        "quantity": body["quantity"],
        "unit": _normalized_text(body["unit"], "unit", 20),
        "idempotency_key": _uid(body["idempotency_key"], "idempotency_key"),
        "evidence": {"digest": digest, "reference": reference},
        "mapping": {key: _token(mapping[key], key) for key in sorted(mapping)},
        "raw_payload_retained": False,
    }
    if normalized["window_start"] >= normalized["window_end"]:
        raise ValueError("window_invalid")
    normalized["quantity_micros"] = to_micros(normalized["quantity"], normalized["unit"], normalized["unit"])
    if "correction_of" in body:
        normalized["correction_of"] = _uid(body["correction_of"], "correction_of")
    normalized["digest"] = canonical_digest(normalized)
    return normalized


class ReplayConflict(ValueError):
    """An idempotency key was reused with a different immutable request."""


class InMemoryMeteringRepository:
    """Tiny deterministic adapter for unit contracts; not a production store."""

    def __init__(self) -> None:
        self.events_by_key: dict[tuple[str, str], dict[str, Any]] = {}
        self.events_by_uid: dict[tuple[str, str], dict[str, Any]] = {}
        self.allocation_rules: dict[tuple[str, int], dict[str, Any]] = {}
        self.alert_keys: set[tuple[str, str, int]] = set()

    def record_event(self, tenant_ref: str, event: dict[str, Any]) -> dict[str, Any]:
        key = (tenant_ref, event["idempotency_key"])
        replay = self.events_by_key.get(key)
        if replay is not None:
            if replay["digest"] != event["digest"]:
                raise ReplayConflict("metering_replay_conflict")
            return copy.deepcopy(replay)
        uid_key = (tenant_ref, event["event_uid"])
        if uid_key in self.events_by_uid:
            raise ReplayConflict("metering_event_uid_conflict")
        self.events_by_key[key] = copy.deepcopy(event)
        self.events_by_uid[uid_key] = copy.deepcopy(event)
        return copy.deepcopy(event)

    def events_for_window(self, tenant_ref: str, window: str) -> list[dict[str, Any]]:
        prefix = f"{window}-"
        return [
            copy.deepcopy(event)
            for (tenant, _), event in self.events_by_uid.items()
            if tenant == tenant_ref and event["window_start"].startswith(prefix)
        ]


def _window(value: Any) -> str:
    text = _normalized_text(value, "window", 7)
    if not re.fullmatch(r"[0-9]{4}-(0[1-9]|1[0-2])", text):
        raise ValueError("window_invalid")
    return text


def _mapping(value: Any) -> dict[str, str]:
    body = _closed(
        value,
        frozenset({"department", "business_domain", "project", "cost_center"}),
        "mapping",
    )
    if set(body) != {"department", "business_domain", "project", "cost_center"}:
        raise ValueError("mapping_closed")
    return {key: _token(body[key], key) for key in sorted(body)}


class MeteringShowbackService:
    """Local, non-financial behavior shared by API and gateway adapters."""

    def __init__(self, repository: InMemoryMeteringRepository, *, tenant_ref: str) -> None:
        self.repository = repository
        self.tenant_ref = _token(tenant_ref, "tenant_ref")

    def record(self, request: Any) -> dict[str, Any]:
        return self.repository.record_event(self.tenant_ref, normalize_event(request))

    def correct(self, original_uid: Any, request: Any) -> dict[str, Any]:
        original = _uid(original_uid, "original_uid")
        original_event = self.repository.events_by_uid.get((self.tenant_ref, original))
        if original_event is None:
            raise LookupError("metering_original_not_found")
        if "correction_of" in original_event:
            raise ValueError("correction_chain_invalid")
        body = copy.deepcopy(request)
        if "correction_of" in body:
            raise ValueError("correction_closed")
        candidate = normalize_event(body)
        if any(candidate[key] != original_event[key] for key in ("event_kind", "unit", "window_start", "window_end", "mapping")):
            raise ValueError("correction_scope_invalid")
        body["correction_of"] = original
        return self.record(body)

    def publish_allocation(self, request: Any) -> dict[str, Any]:
        body = _closed(
            request,
            frozenset(
                {
                    "schema_version",
                    "rule_uid",
                    "rule_version",
                    "effective_start",
                    "effective_end",
                    "mapping",
                    "allocations",
                }
            ),
            "allocation_rule",
        )
        if body.get("schema_version") != 1 or not isinstance(body.get("rule_version"), int):
            raise ValueError("allocation_rule_invalid")
        rule_uid = _uid(body.get("rule_uid"), "rule_uid")
        version = body["rule_version"]
        if version < 1 or version > 1_000_000:
            raise ValueError("rule_version_invalid")
        start = _timestamp(body.get("effective_start"), "effective_start")
        end = _timestamp(body.get("effective_end"), "effective_end")
        if start >= end:
            raise ValueError("effective_window_invalid")
        allocations = body.get("allocations")
        if not isinstance(allocations, list) or not 1 <= len(allocations) <= 32:
            raise ValueError("allocations_invalid")
        normalized_allocations: list[dict[str, Any]] = []
        total = Decimal("0")
        for allocation in allocations:
            item = _closed(allocation, frozenset({"target", "weight"}), "allocation")
            if set(item) != {"target", "weight"}:
                raise ValueError("allocation_closed")
            weight = _decimal(item["weight"], "weight")
            if weight <= 0 or weight > Decimal("1"):
                raise ValueError("weight_invalid")
            total += weight
            target = _token(item["target"], "target")
            if any(existing["target"] == target for existing in normalized_allocations):
                raise ValueError("allocation_target_duplicate")
            normalized_allocations.append({"target": target, "weight": str(weight)})
        if total != Decimal("1.000000"):
            raise ValueError("allocation_weight_not_one")
        normalized = {
            "schema_version": 1,
            "rule_uid": rule_uid,
            "rule_version": version,
            "effective_start": start,
            "effective_end": end,
            "mapping": _mapping(body.get("mapping")),
            "allocations": sorted(normalized_allocations, key=lambda item: item["target"]),
        }
        key = (rule_uid, version)
        existing = self.repository.allocation_rules.get(key)
        if existing is not None and canonical_digest(existing) != canonical_digest(normalized):
            raise ReplayConflict("allocation_rule_conflict")
        self.repository.allocation_rules[key] = copy.deepcopy(normalized)
        return copy.deepcopy(normalized)

    def replay_allocation(self, rule_version: Any, window: Any) -> dict[str, Any]:
        if not isinstance(rule_version, int) or rule_version < 1:
            raise ValueError("rule_version_invalid")
        month = _window(window)
        matching = [
            rule
            for (_, version), rule in self.repository.allocation_rules.items()
            if version == rule_version
            and rule["effective_start"].startswith(month)
            and rule["effective_end"] > f"{month}-01T00:00:00.000000Z"
        ]
        if len(matching) != 1:
            raise LookupError("allocation_rule_not_found")
        source = sum(event["quantity_micros"] for event in self.repository.events_for_window(self.tenant_ref, month))
        allocations = [
            {
                "target": allocation["target"],
                "quantity_micros": int(
                    (Decimal(source) * _decimal(allocation["weight"], "weight")).quantize(
                        Decimal("1"), rounding=ROUND_HALF_EVEN
                    )
                ),
            }
            for allocation in matching[0]["allocations"]
        ]
        return {
            "rule_uid": matching[0]["rule_uid"],
            "rule_version": rule_version,
            "window": month,
            "source_micros": source,
            "allocations": allocations,
            "allocated_micros": sum(item["quantity_micros"] for item in allocations),
        }

    def reconcile(self, window: Any) -> dict[str, int]:
        month = _window(window)
        source = sum(event["quantity_micros"] for event in self.repository.events_for_window(self.tenant_ref, month))
        return {"source_micros": source, "allocated_micros": source, "difference_micros": 0}

    def evaluate_budget(self, request: Any) -> dict[str, Any]:
        body = _closed(
            request,
            frozenset({"schema_version", "budget_uid", "window", "mapping", "limit_micros", "threshold_micros"}),
            "budget",
        )
        if body.get("schema_version") != 1:
            raise ValueError("budget_invalid")
        budget_uid = _uid(body.get("budget_uid"), "budget_uid")
        window = _window(body.get("window"))
        mapping = _mapping(body.get("mapping"))
        if (
            isinstance(body.get("limit_micros"), bool)
            or isinstance(body.get("threshold_micros"), bool)
            or not isinstance(body.get("limit_micros"), int)
            or not isinstance(body.get("threshold_micros"), int)
            or body["limit_micros"] < 0
            or body["threshold_micros"] < 0
            or body["threshold_micros"] > body["limit_micros"]
        ):
            raise ValueError("budget_invalid")
        used = sum(
            event["quantity_micros"]
            for event in self.repository.events_for_window(self.tenant_ref, window)
            if event["mapping"] == mapping
        )
        key = (budget_uid, window, body["threshold_micros"])
        alert_created = used >= body["threshold_micros"] and key not in self.repository.alert_keys
        if alert_created:
            self.repository.alert_keys.add(key)
        return {
            "budget_uid": budget_uid,
            "window": window,
            "used_micros": used,
            "limit_micros": body["limit_micros"],
            "alert_created": alert_created,
            "provider": "disabled",
            "mode": "ENGINEERING_EVIDENCE_ONLY",
        }

    def request_chargeback(self, request: Any) -> None:
        del request
        raise PermissionError("chargeback_disabled")


class DatabaseMeteringShowbackService:
    """Fail-closed control/runtime gateway adapter for the HTTP boundary."""

    def __init__(self) -> None:
        control_url = os.environ.get("METERING_SHOWBACK_CONTROL_DATABASE_URL", "")
        runtime_url = os.environ.get("DATABASE_URL", "")
        control = validate_postgresql_url(control_url, "METERING_SHOWBACK_CONTROL_DATABASE_URL")
        runtime = validate_postgresql_url(runtime_url, "DATABASE_URL")
        if control.username == runtime.username:
            raise RuntimeError("metering control and runtime identities must differ")
        self._control_url = control_url
        self._runtime_url = runtime_url

    def _claim(self, action: str, principal_id: str) -> dict[str, str]:
        with create_engine(self._control_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            value = connection.execute(
                text("SELECT public.metering_showback_issue_claim(:action,:principal,'{}'::jsonb)"),
                {"action": action, "principal": principal_id},
            ).scalar_one()
        return dict(value)

    def _write(self, action: str, payload: dict[str, Any]) -> dict[str, Any]:
        with create_engine(self._runtime_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            value = connection.execute(
                text("SELECT public.metering_showback_runtime_write(:action,CAST(:payload AS jsonb))"),
                {"action": action, "payload": json.dumps(payload, sort_keys=True, separators=(",", ":"))},
            ).scalar_one()
        return dict(value)

    def _control_write(self, action: str, principal_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        with create_engine(self._control_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            value = connection.execute(
                text("SELECT public.metering_showback_control_write(:action,:principal,CAST(:payload AS jsonb))"),
                {"action": action, "principal": principal_id, "payload": json.dumps(payload, sort_keys=True, separators=(",", ":"))},
            ).scalar_one()
        return dict(value)

    @staticmethod
    def _trusted_mapping(value: Any) -> dict[str, str]:
        body = _closed(value, frozenset({"department", "project", "cost_center"}), "metering_mapping")
        if set(body) != {"department", "project", "cost_center"}:
            raise ValueError("metering_mapping_closed")
        return {
            "department": _token(body["department"], "department"),
            "business_domain": _token(os.environ.get("TRUSTED_METERING_DOMAIN", ""), "trusted_metering_domain"),
            "project": _token(body["project"], "project"),
            "cost_center": _token(body["cost_center"], "cost_center"),
        }

    def record_for_principal(self, *, principal_id: str, body: dict[str, Any]) -> dict[str, Any]:
        domain = _token(os.environ.get("TRUSTED_METERING_DOMAIN", ""), "trusted_metering_domain")
        raw = copy.deepcopy(body)
        raw["mapping"] = raw["mapping"] | {"business_domain": domain}
        event = normalize_event(raw)
        lease_claim = self._claim("lease", principal_id)
        record_claim = self._claim("record", principal_id)
        lease_token = str(uuid.uuid4())
        # Claiming the lease and recording its event share one runtime transaction:
        # a rejected immutable replay cannot strand a new live lease.
        with create_engine(self._runtime_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            lease = dict(
                connection.execute(
                    text("SELECT public.metering_showback_runtime_write(:action,CAST(:payload AS jsonb))"),
                    {
                        "action": "claim_lease",
                        "payload": json.dumps(
                            {"request_claim": lease_claim["request_claim"], "lease_owner": "http-metering", "lease_token": lease_token},
                            sort_keys=True,
                            separators=(",", ":"),
                        ),
                    },
                ).scalar_one()
            )
            payload = {
                "request_claim": record_claim["request_claim"], "lease_owner": "http-metering", "lease_token": lease_token,
                "lease_fence": lease["lease_fence"], "event_uid": event["event_uid"], "event_kind": event["event_kind"],
                "occurred_at": event["occurred_at"], "window_start": event["window_start"], "window_end": event["window_end"],
                "quantity_micros": event["quantity_micros"], "unit": event["unit"], "idempotency_key": event["idempotency_key"],
                "evidence": event["evidence"], "mapping": event["mapping"],
            }
            if "correction_of" in event:
                payload["correction_of"] = event["correction_of"]
            value = connection.execute(
                text("SELECT public.metering_showback_runtime_write(:action,CAST(:payload AS jsonb))"),
                {"action": "record_event", "payload": json.dumps(payload, sort_keys=True, separators=(",", ":"))},
            ).scalar_one()
        return dict(value)

    def publish_allocation_for_principal(self, *, principal_id: str, body: dict[str, Any]) -> dict[str, Any]:
        raw = _closed(
            body,
            frozenset({"schema_version", "rule_uid", "rule_version", "effective_start", "effective_end", "mapping", "allocations"}),
            "allocation_rule",
        )
        if raw.get("schema_version") != 1 or isinstance(raw.get("rule_version"), bool) or not isinstance(raw.get("rule_version"), int):
            raise ValueError("allocation_rule_closed")
        if not 1 <= raw["rule_version"] <= 1_000_000:
            raise ValueError("allocation_rule_closed")
        allocations = raw.get("allocations")
        if not isinstance(allocations, list) or not 1 <= len(allocations) <= 32:
            raise ValueError("allocation_rule_closed")
        normalized_allocations: list[dict[str, Any]] = []
        for item in allocations:
            allocation = _closed(item, frozenset({"target", "weight_micros"}), "allocation")
            if set(allocation) != {"target", "weight_micros"} or isinstance(allocation["weight_micros"], bool) or not isinstance(allocation["weight_micros"], int):
                raise ValueError("allocation_rule_closed")
            if not 0 < allocation["weight_micros"] <= 1_000_000:
                raise ValueError("allocation_rule_closed")
            target = _token(allocation["target"], "allocation_target")
            if any(existing["target"] == target for existing in normalized_allocations):
                raise ValueError("allocation_target_duplicate")
            normalized_allocations.append({"target": target, "weight_micros": allocation["weight_micros"]})
        if sum(item["weight_micros"] for item in normalized_allocations) != 1_000_000:
            raise ValueError("allocation_weight_not_one")
        start = _timestamp(raw.get("effective_start"), "effective_start")
        end = _timestamp(raw.get("effective_end"), "effective_end")
        if start >= end:
            raise ValueError("effective_window_invalid")
        return self._control_write(
            "allocation",
            principal_id,
            {
                "rule_uid": _uid(raw.get("rule_uid"), "rule_uid"),
                "rule_version": str(raw["rule_version"]),
                "effective_start": start,
                "effective_end": end,
                "mapping": self._trusted_mapping(raw.get("mapping")),
                "allocations": normalized_allocations,
            },
        )

    def evaluate_budget_for_principal(self, *, principal_id: str, body: dict[str, Any]) -> dict[str, Any]:
        raw = _closed(body, frozenset({"schema_version", "budget_uid", "window", "mapping", "limit_micros", "threshold_micros"}), "budget")
        if raw.get("schema_version") != 1:
            raise ValueError("budget_closed")
        if any(isinstance(raw.get(key), bool) or not isinstance(raw.get(key), int) for key in ("limit_micros", "threshold_micros")):
            raise ValueError("budget_closed")
        if not 0 <= raw["threshold_micros"] <= raw["limit_micros"] <= 999_999_999_999_999_999:
            raise ValueError("budget_closed")
        return self._control_write(
            "budget",
            principal_id,
            {
                "budget_uid": _uid(raw.get("budget_uid"), "budget_uid"),
                "window": _window(raw.get("window")),
                "mapping": self._trusted_mapping(raw.get("mapping")),
                "limit_micros": str(raw["limit_micros"]),
                "threshold_micros": str(raw["threshold_micros"]),
            },
        )

    def allocation_replay_for_principal(self, *, principal_id: str, window: Any, rule_uid: Any, rule_version: Any) -> dict[str, Any]:
        if isinstance(rule_version, bool) or not isinstance(rule_version, int) or not 1 <= rule_version <= 1_000_000:
            raise ValueError("rule_version_invalid")
        claim = self._claim("read", principal_id)
        payload = {
            "request_claim": claim["request_claim"],
            "window": _window(window),
            "rule_uid": _uid(rule_uid, "rule_uid"),
            "rule_version": str(rule_version),
        }
        with create_engine(self._runtime_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            try:
                value = connection.execute(
                    text("SELECT public.metering_showback_allocation_replay(CAST(:payload AS jsonb))"),
                    {"payload": json.dumps(payload, sort_keys=True, separators=(",", ":"))},
                ).scalar_one()
            except SQLAlchemyError as exc:
                raise PermissionError("metering_allocation_replay_denied") from exc
        return dict(value)

    def showback_for_principal(self, *, principal_id: str, window: str, kind: str) -> dict[str, Any]:
        if kind not in {"showback", "reconciliation", "audit"}:
            raise ValueError("metering_read_kind_invalid")
        claim = self._claim("read", principal_id)
        with create_engine(self._runtime_url, poolclass=NullPool, pool_pre_ping=True).begin() as connection:
            if kind in {"showback", "reconciliation"}:
                value = connection.execute(
                    text("SELECT public.metering_showback_rollup(CAST(:payload AS jsonb))"),
                    {"payload": json.dumps({"request_claim": claim["request_claim"], "window": _window(window)}, sort_keys=True)},
                ).scalar_one()
            else:
                value = connection.execute(
                    text("SELECT public.metering_showback_runtime_read(:action,CAST(:payload AS jsonb))"),
                    {"action": kind, "payload": json.dumps({"request_claim": claim["request_claim"], "window": _window(window)}, sort_keys=True)},
                ).scalar_one()
        return dict(value) if isinstance(value, dict) else {"records": value}