"""Connectors: execute manifest extractions against sources (ingest-spec §4).

Pure readers — no logging, no writes (materialization owns the §8 log).
Input-shape failures raise SourceError fail-fast, never a leaked OSError.
"""

from __future__ import annotations

import csv
import os
import sqlite3
from collections.abc import Callable
from contextlib import closing
from pathlib import Path
from urllib.error import URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

from .errors import SourceError
from .render import sql_value_to_text

#: The http transport seam: ``(url, credential) -> response body text``.
#: Injecting a canned implementation in tests keeps the suite socket-free
#: (§11); the default ``urllib_get`` is the ONLY path that opens a socket,
#: and it is reached only inside the gated http branch of materialization.
HttpGet = Callable[[str, "str | None"], str]


def safe_resolve(root: Path, relative: str) -> Path:
    """Resolve `relative` against `root`, fail-closed (the OKF path rule).

    Raises SourceError if the resolved path escapes `root` — `..` traversal,
    an absolute path, a symlink escape, or a prefix-collision sibling
    (`/a/data-evil` vs `/a/data`; commonpath on canonical paths catches what
    a naive startswith would not).
    """
    real_root = os.path.realpath(root)
    try:
        candidate = os.path.realpath(os.path.join(real_root, relative))
    except ValueError as exc:
        # An embedded NUL makes the path syscalls reject the string before
        # any boundary check can run. SourceError promises a typed failure,
        # so a target that cannot be resolved at all fails closed under the
        # boundary code rather than leaking an untyped ValueError.
        raise SourceError(f"path is not resolvable: {relative!r}", code="path_escape") from exc
    try:
        within = os.path.commonpath([real_root, candidate]) == real_root
    except ValueError:
        # Different drives / mixed absolute-relative -> not within.
        within = False
    if not within:
        raise SourceError(f"path escapes its root directory: {relative!r}", code="path_escape")
    return Path(candidate)


def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str], list[list[str]]]:
    """Execute a `file`-source extraction: read the CSV at `query` inside `root`.

    Boundary-checked fail-closed path resolution; utf-8-sig so a BOM never
    leaks into the first header cell; streaming max_rows cap — exceeding it
    is an ERROR the moment it happens, never silent truncation (§8);
    ragged rows rejected, cells returned verbatim (escaping is §5 rendering).
    """
    root_path = Path(root)
    if not root_path.is_dir():
        raise SourceError(
            f"file-source root is not a directory: {root_path}", code="source_root_missing"
        )
    resolved = safe_resolve(root_path, query)
    if not resolved.is_file():
        raise SourceError(
            f"extraction query does not resolve to a file: {query!r}",
            code="source_file_missing",
        )
    with resolved.open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.reader(handle)
        header = next(reader, None)
        if header is None:
            raise SourceError(f"CSV has no header row: {query!r}", code="csv_no_header")
        rows: list[list[str]] = []
        for row in reader:
            if len(rows) >= max_rows:
                raise SourceError(
                    f"extraction {query!r} exceeds max_rows={max_rows} "
                    "(error, never silent truncation — spec §8)",
                    code="max_rows_exceeded",
                )
            if len(row) != len(header):
                raise SourceError(
                    f"ragged CSV row in {query!r}: expected {len(header)} cells, got {len(row)}",
                    code="csv_ragged_row",
                )
            rows.append(row)
    return header, rows


def read_sql(
    connection_ref: str, query: str, *, max_rows: int
) -> tuple[list[str], list[list[str]]]:
    """Execute a `sql`-source extraction: one read-only SELECT against sqlite.

    `connection_ref` names an environment variable whose value is the
    database path — resolved at run time, never from the manifest (§4).
    Opened read-only (mode=ro) so a write in `query` fails at the DB —
    §4's "SHOULD enforce read-only" honoured robustly, not by string
    parsing; Connection.execute runs exactly one statement. Cells are
    converted to their §5 text form; the streaming max_rows cap is an
    ERROR the moment it is exceeded (§8). Env-unset, missing file, and
    any sqlite3 error raise typed errors fail-fast.
    """
    dsn = os.environ.get(connection_ref)
    if not dsn:
        raise SourceError(
            f"sql source connection_ref {connection_ref!r} is not set in the environment "
            "(paths/credentials resolve at run time, never from the manifest — §4)",
            code="connection_ref_unset",
        )
    db_file = Path(dsn)
    if not db_file.is_file():
        raise SourceError(
            f"sql connection_ref {connection_ref!r} points at a missing database file: {db_file}",
            code="database_missing",
        )
    # quote keeps '/' but encodes spaces/'?'/'#' so an odd path can't corrupt
    # the URI; file:{abs path}?mode=ro is sqlite's documented read-only open.
    uri = f"file:{quote(str(db_file))}?mode=ro"
    try:
        with closing(sqlite3.connect(uri, uri=True)) as conn:
            cursor = conn.execute(query)
            if cursor.description is None:  # a SELECT always has columns; defensive
                raise SourceError(
                    f"sql extraction returned no columns: {query!r}", code="sql_no_columns"
                )
            header = [column[0] for column in cursor.description]
            rows: list[list[str]] = []
            for row in cursor:
                if len(rows) >= max_rows:
                    raise SourceError(
                        f"extraction {query!r} exceeds max_rows={max_rows} "
                        "(error, never silent truncation — spec §8)",
                        code="max_rows_exceeded",
                    )
                rows.append([sql_value_to_text(value) for value in row])
    except (sqlite3.Error, sqlite3.Warning) as exc:
        raise SourceError(
            f"sql extraction failed for query {query!r}: {exc}", code="sql_failed"
        ) from exc
    return header, rows


def urllib_get(url: str, credential: str | None) -> str:
    """The stdlib GET — the only socket path in this library.

    Attaches `Authorization: Bearer {credential}` iff a credential was
    resolved. Transport/decode failures wrap as SourceError WITHOUT the
    secret — the credential rides in the header, never in the URL
    (embedded credentials are already rejected at schema validation).
    """
    headers = {"Authorization": f"Bearer {credential}"} if credential is not None else {}
    request = Request(url, headers=headers, method="GET")
    try:
        with urlopen(request) as response:
            raw: bytes = response.read()
        return raw.decode("utf-8")
    except (URLError, OSError, UnicodeDecodeError) as exc:
        raise SourceError(f"http GET failed for {url!r}: {exc}", code="http_transport") from exc


def read_http(
    base_url: str,
    query: str,
    *,
    max_rows: int,
    credential_ref: str | None = None,
    get: HttpGet = urllib_get,
) -> str:
    """Execute an `http`-source extraction: GET `base_url` joined with `query`
    via the injectable `get` seam, returning the body for verbatim fenced
    rendering (§4/§5).

    `credential_ref` names an environment variable holding the secret,
    resolved at run time (present-but-unset is fail-fast, before any
    transport call; None means no auth) — never read from the manifest,
    never logged, never stamped. URL join is explicit (one slash), not
    urljoin. The body is CRLF→LF normalized and capped on its LF line count
    (a missing trailing newline still counts the last line); exceeding
    max_rows is an ERROR (§8). A body line that is a code-fence marker is
    rejected — it cannot be safely embedded in a fenced block; fail-fast,
    never silent corruption.
    """
    if credential_ref is None:
        credential: str | None = None
    else:
        credential = os.environ.get(credential_ref)
        if not credential:
            raise SourceError(
                f"http source credential_ref {credential_ref!r} is not set in the environment "
                "(credentials resolve at run time, never from the manifest — §4)",
                code="credential_ref_unset",
            )
    url = base_url.rstrip("/") + "/" + query.lstrip("/")
    normalized = get(url, credential).replace("\r\n", "\n").replace("\r", "\n")
    for line in normalized.split("\n"):
        if line.lstrip(" ").startswith("```"):
            raise SourceError(
                f"http extraction {query!r} contains a code-fence marker line (```) — "
                "cannot be safely embedded in a fenced code block; fail-fast, never "
                "silent corruption",
                code="fence_marker_in_body",
            )
    newline_count = normalized.count("\n")
    line_count = (
        newline_count if (normalized == "" or normalized.endswith("\n")) else newline_count + 1
    )
    if line_count > max_rows:
        raise SourceError(
            f"http extraction {query!r} exceeds max_rows={max_rows} "
            "(error, never silent truncation — spec §8)",
            code="max_rows_exceeded",
        )
    return normalized