# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/over-what-memory-and-streaming-bounds-must-an-instance-chunk-unbounded-data-ingestions-to-prevent-in-memory-materialization-exhaustion.md
# Remnant Warrant:    https://corpusfetch.com/c/over-what-memory-and-streaming-bounds-must-an-instance-chunk-unbounded-data-ingestions-to-prevent-in-memory-materialization-exhaustion.warrant.json
# Markdown SHA-256:   9042e74277f6952e1c02a1d5b48f739e4869d21838ad487a75eff019a996dda0
# Code AST SHA-256:   463df116b45fa84404b173c2749ee4bd4090c39b525b60cf97eadbd6eb135c77
# Verified Classes:   UnboundedMemoryMaterializationError, StreamIngestionReceipt, BoundedStreamIngestor
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import ast
import hashlib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import BinaryIO, Callable, Final


class UnboundedMemoryMaterializationError(MemoryError):
    """Raised when code attempts an unbounded eager read or when a stream exceeds its byte ceiling."""
    pass


@dataclass(frozen=True)
class StreamIngestionReceipt:
    total_chunks_processed: int
    total_bytes_processed: int
    peak_chunk_buffer_bytes: int
    stream_sha256: str


class BoundedStreamIngestor:
    """
    Prevents Out-Of-Memory (OOM) exhaustion and decompression bombs during data ingestion:
    1. Audits Python ASTs for un-argumented `.read()`, `.readlines()`, and `.fetchall()` calls.
    2. Checks `os.stat(path).st_size` before permitting any eager read.
    3. Streams files/streams in fixed `chunk_size_bytes` (`O(1)` memory) while enforcing
       a hard `max_total_stream_bytes` decompression/ingestion ceiling.
    """

    DEFAULT_CHUNK_BYTES: Final[int] = 64 * 1024          # 64 KB constant working memory
    MAX_EAGER_READ_BYTES: Final[int] = 16 * 1024 * 1024  # 16 MB eager ceiling
    FORBIDDEN_UNBOUNDED_METHODS: Final[frozenset[str]] = frozenset({
        "readlines", "read_text", "read_bytes", "fetchall"
    })

    @classmethod
    def assert_no_unbounded_reads_in_ast(cls, source_code: str) -> None:
        tree = ast.parse(source_code)
        for node in ast.walk(tree):
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                method = node.func.attr
                if method in cls.FORBIDDEN_UNBOUNDED_METHODS and len(node.args) == 0:
                    raise UnboundedMemoryMaterializationError(
                        f"Line {node.lineno}: Unbounded eager call '.{method}()' is prohibited on uninspected "
                        "inputs. Use chunked iteration or pass an explicit size bound."
                    )
                if method == "read" and len(node.args) == 0 and len(node.keywords) == 0:
                    raise UnboundedMemoryMaterializationError(
                        f"Line {node.lineno}: Unbounded '.read()' without a byte limit argument is prohibited."
                    )

    @classmethod
    def stream_with_ceiling(
        cls,
        stream: BinaryIO,
        *,
        on_chunk: Callable[[bytes], None],
        chunk_size_bytes: int = DEFAULT_CHUNK_BYTES,
        max_total_stream_bytes: int = 512 * 1024 * 1024,
    ) -> StreamIngestionReceipt:
        if chunk_size_bytes <= 0 or chunk_size_bytes > cls.MAX_EAGER_READ_BYTES:
            raise ValueError(f"Invalid chunk_size_bytes: {chunk_size_bytes}")

        hasher = hashlib.sha256()
        total_bytes = 0
        chunks = 0
        peak_buf = 0

        while True:
            chunk = stream.read(chunk_size_bytes)
            if not chunk:
                break

            total_bytes += len(chunk)
            if total_bytes > max_total_stream_bytes:
                raise UnboundedMemoryMaterializationError(
                    f"Stream / Decompression Ceiling Exceeded: ingested {total_bytes} bytes "
                    f"(limit={max_total_stream_bytes} bytes). Aborting to prevent OOM."
                )

            chunks += 1
            peak_buf = max(peak_buf, len(chunk))
            hasher.update(chunk)
            on_chunk(chunk)

        return StreamIngestionReceipt(
            total_chunks_processed=chunks,
            total_bytes_processed=total_bytes,
            peak_chunk_buffer_bytes=peak_buf,
            stream_sha256=hasher.hexdigest(),
        )

    @classmethod
    def safe_read_small_file(cls, path: Path, *, max_bytes: int = MAX_EAGER_READ_BYTES) -> bytes:
        st = os.stat(path)
        if st.st_size > max_bytes:
            raise UnboundedMemoryMaterializationError(
                f"File '{path}' is {st.st_size} bytes, exceeding eager read ceiling ({max_bytes} bytes). "
                "Use stream_with_ceiling() instead."
            )
        with path.open("rb") as f:
            return f.read(max_bytes)