"""Intra- and inter-procedural dataflow heuristics.

This module implements a lightweight, deterministic subset:
- dead assignments (assigned variable never read in the same function)
- unused parameters (parameter never read in function body)
- source-to-sink flow in the same function (string-pattern based)
- inter_source_to_sink (cross-function taint from taint analysis tables)
- inter_unused_param (params with no dataflow effect per taint summaries)
- inter_unused_return (return values not consumed by callers)

It is designed for fast, index-backed scans and custom `dataflow_match` rules.
"""

from __future__ import annotations

import fnmatch
import json
import re
import sqlite3
from collections import Counter
from pathlib import Path

from roam._signature_utils import parse_param_names as _parse_param_names
from roam.db.connection import batched_in, find_project_root
from roam.db.edge_kinds import CALL_EDGE_KINDS

_IDENT_RE = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\b")

_ASSIGN_PATTERNS = (
    re.compile(r"^\s*(?:let|const|var)\s+([A-Za-z_][A-Za-z0-9_]*)"),
    re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=(?!=)"),
    re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*[\+\-\*/%]="),
)

_DEFAULT_SOURCES = (
    "input(",
    "request.args",
    "request.form",
    "request.get",
    "request.GET",
    "request.POST",
    "sys.argv",
    "os.environ",
    "query_params",
    "params.get",
)

_DEFAULT_SINKS = (
    "eval(",
    "exec(",
    "os.system(",
    "subprocess.run(",
    "subprocess.popen(",
    "pickle.loads(",
    "yaml.load(",
    "Function(",
)

# Names ignored in local dataflow scans (assignment + unused-param
# detectors). ``_parse_param_names`` keeps its own ignore set in
# ``roam._signature_utils``; this one stays local because the dataflow
# detectors here also filter ``_``-prefixed names via ``startswith``,
# which the canonical param-name parser deliberately does not.
_IGNORED_NAMES = {"_", "self", "cls"}


def _match_glob(path: str, pattern: str | None) -> bool:
    if not pattern:
        return True
    norm = (path or "").replace("\\", "/")
    pat = pattern.replace("\\", "/")
    if fnmatch.fnmatch(norm, pat):
        return True
    return Path(norm).match(pat)


# W886: delegate to the canonical commands-layer helper (W873 audit).
# The local variant covered the JS spec-suffix case (``.spec.js``/
# ``.spec.ts``) but missed Vue SFC specs (``.spec.vue``), Go
# ``*_test.go``, PHP ``*Test.php``, ``__tests__/`` / ``spec/`` /
# ``testing/`` directories, and ``conftest.py``. Canonical helper
# covers all of those via ``test_conventions.is_test_file``.
from roam.commands.changed_files import is_test_file as _is_test_path

# ``_parse_param_names`` is imported from ``roam._signature_utils`` at
# the top of this module (W856 hoist — was duplicated in
# ``analysis/taint.py``).


def _read_file_lines(project_root: Path, rel_path: str) -> list[str]:
    path = project_root / rel_path
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    return text.splitlines()


def _extract_assignments(body_lines: list[str], start_line: int) -> list[tuple[str, int]]:
    out: list[tuple[str, int]] = []
    for idx, line in enumerate(body_lines):
        # Strip trailing comments for basic signal quality.
        line_no_comments = line.split("#", 1)[0].split("//", 1)[0]
        for pattern in _ASSIGN_PATTERNS:
            m = pattern.search(line_no_comments)
            if not m:
                continue
            name = m.group(1).strip()
            if not name or name in _IGNORED_NAMES:
                continue
            out.append((name, start_line + idx))
            break
    return out


def _token_counts(text: str) -> Counter:
    return Counter(match.group(1) for match in _IDENT_RE.finditer(text))


def _find_source_sink(
    body_lines: list[str],
    start_line: int,
    *,
    sources: tuple[str, ...],
    sinks: tuple[str, ...],
) -> tuple[int, str, int, str] | None:
    source_hit: tuple[int, str] | None = None
    sink_hit: tuple[int, str] | None = None

    for idx, raw in enumerate(body_lines):
        line = raw.lower()
        line_no = start_line + idx
        if source_hit is None:
            for src in sources:
                if src.lower() in line:
                    source_hit = (line_no, src)
                    break
        for sink in sinks:
            if sink.lower() in line:
                sink_hit = (line_no, sink)
                break
        if source_hit and sink_hit and source_hit[0] <= sink_hit[0]:
            return source_hit[0], source_hit[1], sink_hit[0], sink_hit[1]
    return None


_ALL_PATTERNS = frozenset(
    {
        "dead_assignment",
        "unused_param",
        "source_to_sink",
        "inter_source_to_sink",
        "inter_unused_param",
        "inter_unused_return",
    }
)

_INTRA_DEFAULTS = frozenset({"dead_assignment", "unused_param", "source_to_sink"})

_INTER_PATTERNS = frozenset({"inter_source_to_sink", "inter_unused_param", "inter_unused_return"})


def _normalize_patterns(patterns: list[str] | tuple[str, ...] | str | None) -> set[str]:
    if patterns is None:
        return set(_INTRA_DEFAULTS)
    if isinstance(patterns, str):
        values = [patterns]
    else:
        values = list(patterns)
    out = set()
    for value in values:
        key = str(value).strip().lower()
        if key in _ALL_PATTERNS:
            out.add(key)
    return out


def _table_available(conn, *names: str) -> bool:
    """Cheap existence check on every table in ``names``."""
    if not names:
        return True
    rows = batched_in(
        conn,
        "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ({ph})",
        names,
    )
    return sum(row[0] for row in rows) == len(names)


def _inter_source_to_sink_findings(conn, file_glob: str | None) -> list[dict]:
    """Cross-function taint flows: a tainted source reaches a sink in
    another function. Reads from ``taint_findings``."""
    rows = conn.execute(
        """
        SELECT tf.source_type, tf.sink_type, tf.call_chain,
               tf.chain_length, tf.confidence,
               s1.name AS src_name, COALESCE(s1.qualified_name, s1.name) AS src_qname,
               f1.path AS src_file, s1.line_start AS src_line,
               s2.name AS sink_name, COALESCE(s2.qualified_name, s2.name) AS sink_qname,
               f2.path AS sink_file, s2.line_start AS sink_line
        FROM taint_findings tf
        JOIN symbols s1 ON tf.source_symbol_id = s1.id
        JOIN files f1 ON s1.file_id = f1.id
        JOIN symbols s2 ON tf.sink_symbol_id = s2.id
        JOIN files f2 ON s2.file_id = f2.id
        WHERE tf.sanitized = 0
        ORDER BY tf.confidence DESC
        """
    ).fetchall()
    out: list[dict] = []
    for row in rows:
        src_file = (row["src_file"] or "").replace("\\", "/")
        sink_file = (row["sink_file"] or "").replace("\\", "/")
        if file_glob and not _match_glob(src_file, file_glob) and not _match_glob(sink_file, file_glob):
            continue
        out.append(
            {
                "type": "inter_source_to_sink",
                "symbol": row["src_qname"],
                "file": src_file,
                "line": row["src_line"],
                "sink_symbol": row["sink_qname"],
                "sink_file": sink_file,
                "sink_line": row["sink_line"],
                "source": row["source_type"],
                "sink": row["sink_type"],
                "chain_length": row["chain_length"],
                "confidence": row["confidence"],
                "reason": "cross-function taint: {} in {} -> {} in {} (depth {})".format(
                    row["source_type"],
                    row["src_qname"],
                    row["sink_type"],
                    row["sink_qname"],
                    row["chain_length"],
                ),
            }
        )
    return out


def _inter_unused_param_findings(conn, file_glob: str | None) -> list[dict]:
    """Parameters that don't influence the return value or any sink."""
    if not _table_available(conn, "taint_summaries"):
        return []
    rows = conn.execute(
        """
        SELECT ts.symbol_id, ts.param_taints_return, ts.param_to_sink,
               s.name, COALESCE(s.qualified_name, s.name) AS qname,
               s.signature, f.path AS file_path, s.line_start
        FROM taint_summaries ts
        JOIN symbols s ON ts.symbol_id = s.id
        JOIN files f ON s.file_id = f.id
        WHERE ts.is_sanitizer = 0
        """
    ).fetchall()
    out: list[dict] = []
    for row in rows:
        fp = (row["file_path"] or "").replace("\\", "/")
        if file_glob and not _match_glob(fp, file_glob):
            continue
        try:
            ptr = json.loads(row["param_taints_return"] or "{}")
            pts = json.loads(row["param_to_sink"] or "{}")
        except json.JSONDecodeError:
            continue
        params = _parse_param_names(row["signature"])
        for idx, pname in enumerate(params):
            sidx = str(idx)
            if not ptr.get(sidx, False) and not pts.get(sidx):
                out.append(
                    {
                        "type": "inter_unused_param",
                        "symbol": row["qname"],
                        "file": fp,
                        "line": row["line_start"],
                        "variable": pname,
                        "reason": "parameter '{}' in {} has no dataflow effect (not in return, not in sink)".format(
                            pname, row["qname"]
                        ),
                    }
                )
    return out


def _inter_unused_return_findings(conn, file_glob: str | None) -> list[dict]:
    """Return values that are computed but never reach a source."""
    if not _table_available(conn, "taint_summaries", "symbol_metrics"):
        return []
    rows = conn.execute(
        """
        SELECT ts.symbol_id, s.name, COALESCE(s.qualified_name, s.name) AS qname,
               f.path AS file_path, s.line_start
        FROM taint_summaries ts
        JOIN symbols s ON ts.symbol_id = s.id
        JOIN files f ON s.file_id = f.id
        JOIN symbol_metrics sm ON s.id = sm.symbol_id
        WHERE sm.return_count > 0
          AND ts.return_from_source = 0
          AND ts.is_sanitizer = 0
        """
    ).fetchall()
    out: list[dict] = []
    for row in rows:
        fp = (row["file_path"] or "").replace("\\", "/")
        if file_glob and not _match_glob(fp, file_glob):
            continue
        # W512: edge-kind vocabulary lives in roam.db.edge_kinds. Pure
        # call edges only — unused-return is a call-site concept.
        _df_call_kind_ph = ", ".join("?" for _ in CALL_EDGE_KINDS)
        caller_count = conn.execute(
            f"SELECT COUNT(*) FROM edges WHERE target_id = ? AND kind IN ({_df_call_kind_ph})",
            (row["symbol_id"], *CALL_EDGE_KINDS),
        ).fetchone()[0]
        if caller_count > 0:
            out.append(
                {
                    "type": "inter_unused_return",
                    "symbol": row["qname"],
                    "file": fp,
                    "line": row["line_start"],
                    "reason": "return value of {} is computed but may not be used by callers".format(row["qname"]),
                }
            )
    return out


def _collect_inter_findings(
    conn,
    *,
    patterns: set[str],
    file_glob: str | None = None,
    max_matches: int = 0,
) -> list[dict]:
    """Collect inter-procedural dataflow findings from taint analysis tables."""
    if not _table_available(conn, "taint_findings"):
        return []

    findings: list[dict] = []
    if "inter_source_to_sink" in patterns:
        findings.extend(_inter_source_to_sink_findings(conn, file_glob))
    if "inter_unused_param" in patterns:
        findings.extend(_inter_unused_param_findings(conn, file_glob))
    if "inter_unused_return" in patterns:
        findings.extend(_inter_unused_return_findings(conn, file_glob))

    if max_matches > 0:
        findings = findings[:max_matches]
    return findings


def _collect_intra_findings_for_symbol(
    row: sqlite3.Row,
    *,
    target_patterns: set[str],
    file_path: str,
    symbol_name: str,
    line_start: int,
    body_lines: list[str],
    source_patterns: tuple[str, ...],
    sink_patterns: tuple[str, ...],
) -> list[dict]:
    """Collect intra-procedural findings for one function or method body."""
    if not body_lines:
        return []

    body_text = "\n".join(body_lines)
    token_counts = _token_counts(body_text)
    findings: list[dict] = []

    if "dead_assignment" in target_patterns:
        seen_vars: set[str] = set()
        assignments = _extract_assignments(body_lines, line_start + 1)
        for var_name, var_line in assignments:
            if var_name in seen_vars:
                continue
            seen_vars.add(var_name)
            if token_counts.get(var_name, 0) <= 1:
                findings.append(
                    {
                        "type": "dead_assignment",
                        "symbol": symbol_name,
                        "file": file_path,
                        "line": var_line,
                        "variable": var_name,
                        "reason": (f"'{var_name}' is assigned but never read in {symbol_name}"),
                    }
                )

    if "unused_param" in target_patterns:
        params = _parse_param_names(row["signature"])
        for name in params:
            if name in _IGNORED_NAMES or name.startswith("_"):
                continue
            if token_counts.get(name, 0) <= 0:
                findings.append(
                    {
                        "type": "unused_param",
                        "symbol": symbol_name,
                        "file": file_path,
                        "line": line_start,
                        "variable": name,
                        "reason": (f"parameter '{name}' is never read in {symbol_name}"),
                    }
                )

    if "source_to_sink" in target_patterns:
        hit = _find_source_sink(
            body_lines,
            line_start + 1,
            sources=source_patterns,
            sinks=sink_patterns,
        )
        if hit is not None:
            src_line, source, sink_line, sink = hit
            findings.append(
                {
                    "type": "source_to_sink",
                    "symbol": symbol_name,
                    "file": file_path,
                    "line": sink_line,
                    "source_line": src_line,
                    "source": source,
                    "sink": sink,
                    "reason": (f"possible source-to-sink flow in {symbol_name}: '{source}' -> '{sink}'"),
                }
            )

    return findings


def _body_context_when_scope_allows_dataflow(
    row: sqlite3.Row,
    *,
    file_glob: str | None,
    project_root: Path,
    file_cache: dict[str, list[str]],
) -> tuple[str, str, int, list[str]] | None:
    """Return the scoped body slice for one symbol, or ``None`` when skipped."""
    file_path = (row["file_path"] or "").replace("\\", "/")
    if not file_path or _is_test_path(file_path):
        return None
    if not _match_glob(file_path, file_glob):
        return None

    lines = file_cache.get(file_path)
    if lines is None:
        lines = _read_file_lines(project_root, file_path)
        file_cache[file_path] = lines
    if not lines:
        return None

    line_start = max(int(row["line_start"] or 1), 1)
    line_end = int(row["line_end"] or line_start)
    if line_end < line_start:
        line_end = line_start
    if line_start > len(lines):
        return None

    body_lines = lines[line_start : min(line_end, len(lines))]
    symbol_name = row["qualified_name"] or row["name"]
    return file_path, symbol_name, line_start, body_lines


def _sort_findings_for_stable_budget(findings: list[dict], max_matches: int) -> list[dict]:
    findings.sort(key=lambda f: (f["file"], int(f.get("line") or 0), f["type"], f.get("symbol") or ""))
    if max_matches > 0:
        return findings[:max_matches]
    return findings


def _collect_intra_findings_until_budget(
    rows: list[sqlite3.Row],
    *,
    target_patterns: set[str],
    file_glob: str | None,
    max_matches: int,
    source_patterns: tuple[str, ...],
    sink_patterns: tuple[str, ...],
) -> list[dict]:
    project_root = find_project_root()
    file_cache: dict[str, list[str]] = {}
    findings: list[dict] = []

    for row in rows:
        body_context = _body_context_when_scope_allows_dataflow(
            row,
            file_glob=file_glob,
            project_root=project_root,
            file_cache=file_cache,
        )
        if body_context is None:
            continue

        file_path, symbol_name, line_start, body_lines = body_context
        findings.extend(
            _collect_intra_findings_for_symbol(
                row,
                target_patterns=target_patterns,
                file_path=file_path,
                symbol_name=symbol_name,
                line_start=line_start,
                body_lines=body_lines,
                source_patterns=source_patterns,
                sink_patterns=sink_patterns,
            )
        )

        if max_matches > 0 and len(findings) >= max_matches:
            break

    return findings


def collect_dataflow_findings(
    conn,
    *,
    patterns: list[str] | tuple[str, ...] | str | None = None,
    file_glob: str | None = None,
    max_matches: int = 0,
    sources: list[str] | tuple[str, ...] | None = None,
    sinks: list[str] | tuple[str, ...] | None = None,
) -> list[dict]:
    """Collect intra- and inter-procedural dataflow findings."""
    target_patterns = _normalize_patterns(patterns)
    if not target_patterns:
        return []

    # Collect inter-procedural findings from taint tables
    inter_patterns = target_patterns & _INTER_PATTERNS
    inter_findings: list[dict] = []
    if inter_patterns:
        inter_findings = _collect_inter_findings(
            conn,
            patterns=inter_patterns,
            file_glob=file_glob,
            max_matches=max_matches,
        )

    # If only inter-procedural patterns requested, return early
    intra_patterns = target_patterns - _INTER_PATTERNS
    if not intra_patterns:
        return _sort_findings_for_stable_budget(inter_findings, max_matches)

    source_patterns = tuple(sources) if sources else _DEFAULT_SOURCES
    sink_patterns = tuple(sinks) if sinks else _DEFAULT_SINKS

    rows = conn.execute(
        """
        SELECT
            s.id,
            s.name,
            COALESCE(s.qualified_name, s.name) AS qualified_name,
            s.signature,
            s.kind,
            s.line_start,
            s.line_end,
            f.path AS file_path
        FROM symbols s
        JOIN files f ON s.file_id = f.id
        WHERE s.kind IN ('function', 'method')
          AND s.line_start IS NOT NULL
          AND s.line_end IS NOT NULL
          AND s.line_end >= s.line_start
        ORDER BY f.path, s.line_start
        """
    ).fetchall()

    findings = _collect_intra_findings_until_budget(
        rows,
        target_patterns=intra_patterns,
        file_glob=file_glob,
        max_matches=max_matches,
        source_patterns=source_patterns,
        sink_patterns=sink_patterns,
    )

    # Merge inter-procedural findings
    findings.extend(inter_findings)

    return _sort_findings_for_stable_budget(findings, max_matches)
