wyndham-ARR/arr_storage/contracts.py

"""Provider-neutral contracts for private ARR object storage."""

from __future__ import annotations

import re
from dataclasses import dataclass
from pathlib import PurePosixPath
from typing import BinaryIO, Mapping, Optional, Protocol

from arr_ingestion.contracts import (
    ARTIFACT_SIZE_LIMITS,
    OPAQUE_ID_RE,
    ROLE_CONTRACTS,
    ArtifactRef,
    IngestionError,
)


OBJECT_METADATA_SCHEMA = "1.0"
OBJECT_STATES = frozenset({"staged", "committed"})
CANONICAL_OBJECT_FILENAMES = {
    "source_xml": "source.xml",
    "booking_source": "booking-source.xlsx",
    "daily_report": "daily-report.xlsx",
    "monthly_report": "monthly-report.xlsx",
    "company_report": "company-report.xlsx",
    "result_json": "result.json",
    "structured_result_json": "structured-result.json",
    "exception_report": "exception-report.xlsx",
}
PREFIX_SEGMENT_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$")


def valid_object_key(value: object) -> bool:
    if not isinstance(value, str) or not value or len(value) > 1024:
        return False
    if (
        value.startswith("/")
        or value.endswith("/")
        or "\\" in value
        or "//" in value
        or any(ord(char) < 32 or ord(char) == 127 for char in value)
    ):
        return False
    path = PurePosixPath(value)
    return not path.is_absolute() and "." not in path.parts and ".." not in path.parts


def valid_delivery_filename(role: str, value: object) -> bool:
    if role not in ROLE_CONTRACTS or not isinstance(value, str):
        return False
    _file_kind, extension, _mime_type = ROLE_CONTRACTS[role]
    return (
        0 < len(value) <= 255
        and PurePosixPath(value).name == value
        and value not in {".", ".."}
        and "/" not in value
        and "\\" not in value
        and not any(ord(char) < 32 or ord(char) == 127 for char in value)
        and value.lower().endswith(extension)
    )


@dataclass(frozen=True)
class ObjectAddress:
    job_id: str
    attempt_no: int
    state: str
    role: str


class ObjectKeyPolicy:
    """Builds and parses the only accepted ARR processing object-key shape."""

    def __init__(self, prefix: str = "arr") -> None:
        parts = tuple(prefix.split("/"))
        if not parts or any(not PREFIX_SEGMENT_RE.fullmatch(part) for part in parts):
            raise ValueError("object prefix is invalid")
        self._parts = parts
        self.prefix = "/".join(parts)

    def build(self, address: ObjectAddress) -> str:
        self._validate_address(address)
        filename = CANONICAL_OBJECT_FILENAMES[address.role]
        return "/".join(
            self._parts
            + (
                "jobs",
                address.job_id,
                "attempts",
                f"{address.attempt_no:04d}",
                address.state,
                address.role,
                filename,
            )
        )

    def parse(self, object_key: str) -> ObjectAddress:
        if not valid_object_key(object_key):
            raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid")
        parts = tuple(object_key.split("/"))
        tail = parts[len(self._parts) :]
        if parts[: len(self._parts)] != self._parts or len(tail) != 7:
            raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid")
        jobs, job_id, attempts, attempt_text, state, role, filename = tail
        if jobs != "jobs" or attempts != "attempts" or not attempt_text.isdigit():
            raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid")
        attempt_no = int(attempt_text)
        address = ObjectAddress(job_id, attempt_no, state, role)
        self._validate_address(address)
        if attempt_text != f"{attempt_no:04d}" or filename != CANONICAL_OBJECT_FILENAMES[role]:
            raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object key is invalid")
        return address

    @staticmethod
    def _validate_address(address: ObjectAddress) -> None:
        if (
            not isinstance(address.job_id, str)
            or not OPAQUE_ID_RE.fullmatch(address.job_id)
            or not isinstance(address.attempt_no, int)
            or isinstance(address.attempt_no, bool)
            or not 1 <= address.attempt_no <= 9999
            or address.state not in OBJECT_STATES
            or address.role not in CANONICAL_OBJECT_FILENAMES
        ):
            raise IngestionError("ARTIFACT_REFERENCE_INVALID", "object address is invalid")


@dataclass(frozen=True)
class BackendObject:
    object_key: str
    byte_size: int
    metadata: Mapping[str, str]
    etag: Optional[str] = None
    version_id: Optional[str] = None


class BackendError(RuntimeError):
    """Normalized backend failure that never includes provider response text."""

    def __init__(self, kind: str):
        super().__init__(kind)
        self.kind = kind


class ObjectBackend(Protocol):
    """Adapter seam implemented by a filesystem, OSS, or S3-compatible client."""

    def put_file(
        self,
        object_key: str,
        source: str,
        mime_type: str,
        metadata: Mapping[str, str],
        *,
        if_absent: bool,
    ) -> BackendObject:
        ...

    def head(self, object_key: str) -> BackendObject:
        ...

    def open_reader(self, object_key: str) -> BinaryIO:
        ...

    def copy_object(
        self,
        source_key: str,
        destination_key: str,
        metadata: Mapping[str, str],
        *,
        if_absent: bool,
    ) -> BackendObject:
        ...

    def delete_object(self, object_key: str) -> None:
        ...


@dataclass(frozen=True)
class StoredObject:
    object_key: str
    job_id: str
    attempt_no: int
    state: str
    role: str
    original_filename: str
    sha256: str
    byte_size: int
    mime_type: str
    etag: Optional[str] = None
    version_id: Optional[str] = None

    def to_artifact_ref(self) -> ArtifactRef:
        if self.state != "committed":
            raise IngestionError(
                "ARTIFACT_NOT_COMMITTED", "only committed objects may enter a delivery"
            )
        file_kind, _extension, _mime_type = ROLE_CONTRACTS[self.role]
        return ArtifactRef(
            role=self.role,
            file_kind=file_kind,
            object_key=self.object_key,
            original_filename=self.original_filename,
            sha256=self.sha256,
            byte_size=self.byte_size,
            mime_type=self.mime_type,
        )


def role_limit(role: str) -> int:
    try:
        return ARTIFACT_SIZE_LIMITS[role]
    except KeyError:
        raise IngestionError("ARTIFACT_REFERENCE_INVALID", "artifact role is invalid") from None