rose-ash/shared/sx/deps.py

"""
Component dependency analysis.

Thin host wrapper over bootstrapped deps module from shared/sx/ref/deps.sx.
The canonical logic lives in the spec; this module provides Python-typed
entry points for the rest of the codebase.
"""
from __future__ import annotations

import os
from typing import Any

from .types import Component, Macro, Symbol


def _use_ref() -> bool:
    return os.environ.get("SX_USE_REF") == "1"


# ---------------------------------------------------------------------------
# Hand-written fallback (used when SX_USE_REF != 1)
# ---------------------------------------------------------------------------

def _scan_ast(node: Any) -> set[str]:
    refs: set[str] = set()
    _walk(node, refs)
    return refs


def _walk(node: Any, refs: set[str]) -> None:
    if isinstance(node, Symbol):
        if node.name.startswith("~"):
            refs.add(node.name)
        return
    if isinstance(node, list):
        for item in node:
            _walk(item, refs)
        return
    if isinstance(node, dict):
        for v in node.values():
            _walk(v, refs)
        return


def _transitive_deps_fallback(name: str, env: dict[str, Any]) -> set[str]:
    seen: set[str] = set()

    def walk(n: str) -> None:
        if n in seen:
            return
        seen.add(n)
        val = env.get(n)
        if isinstance(val, Component):
            for dep in _scan_ast(val.body):
                walk(dep)
        elif isinstance(val, Macro):
            for dep in _scan_ast(val.body):
                walk(dep)

    key = name if name.startswith("~") else f"~{name}"
    walk(key)
    return seen - {key}


def _compute_all_deps_fallback(env: dict[str, Any]) -> None:
    for key, val in env.items():
        if isinstance(val, Component):
            val.deps = _transitive_deps_fallback(key, env)


def _scan_io_refs_fallback(node: Any, io_names: set[str]) -> set[str]:
    """Scan an AST node for references to IO primitive names."""
    refs: set[str] = set()
    _walk_io(node, io_names, refs)
    return refs


def _walk_io(node: Any, io_names: set[str], refs: set[str]) -> None:
    if isinstance(node, Symbol):
        if node.name in io_names:
            refs.add(node.name)
        return
    if isinstance(node, list):
        for item in node:
            _walk_io(item, io_names, refs)
        return
    if isinstance(node, dict):
        for v in node.values():
            _walk_io(v, io_names, refs)
        return


def _transitive_io_refs_fallback(
    name: str, env: dict[str, Any], io_names: set[str]
) -> set[str]:
    """Compute transitive IO primitive references for a component."""
    all_refs: set[str] = set()
    seen: set[str] = set()

    def walk(n: str) -> None:
        if n in seen:
            return
        seen.add(n)
        val = env.get(n)
        if isinstance(val, Component):
            all_refs.update(_scan_io_refs_fallback(val.body, io_names))
            for dep in _scan_ast(val.body):
                walk(dep)
        elif isinstance(val, Macro):
            all_refs.update(_scan_io_refs_fallback(val.body, io_names))
            for dep in _scan_ast(val.body):
                walk(dep)

    key = name if name.startswith("~") else f"~{name}"
    walk(key)
    return all_refs


def _compute_all_io_refs_fallback(
    env: dict[str, Any], io_names: set[str]
) -> None:
    for key, val in env.items():
        if isinstance(val, Component):
            val.io_refs = _transitive_io_refs_fallback(key, env, io_names)


def _scan_components_from_sx_fallback(source: str) -> set[str]:
    import re
    return {f"~{m}" for m in re.findall(r'\(~([a-zA-Z_][a-zA-Z0-9_\-]*)', source)}


def _components_needed_fallback(page_sx: str, env: dict[str, Any]) -> set[str]:
    direct = _scan_components_from_sx_fallback(page_sx)
    all_needed: set[str] = set()
    for name in direct:
        all_needed.add(name)
        val = env.get(name)
        if isinstance(val, Component) and val.deps:
            all_needed.update(val.deps)
        else:
            all_needed.update(_transitive_deps_fallback(name, env))
    return all_needed


# ---------------------------------------------------------------------------
# Public API — dispatches to bootstrapped or fallback
# ---------------------------------------------------------------------------

def transitive_deps(name: str, env: dict[str, Any]) -> set[str]:
    """Compute transitive component dependencies for *name*.

    Returns the set of all component names (with ~ prefix) that
    *name* can transitively render, NOT including *name* itself.
    """
    if _use_ref():
        from .ref.sx_ref import transitive_deps as _ref_td
        return set(_ref_td(name, env))
    return _transitive_deps_fallback(name, env)


def compute_all_deps(env: dict[str, Any]) -> None:
    """Compute and cache deps for all Component entries in *env*."""
    if _use_ref():
        from .ref.sx_ref import compute_all_deps as _ref_cad
        _ref_cad(env)
        return
    _compute_all_deps_fallback(env)


def scan_components_from_sx(source: str) -> set[str]:
    """Extract component names referenced in SX source text.

    Returns names with ~ prefix, e.g. {"~card", "~nav-link"}.
    """
    if _use_ref():
        from .ref.sx_ref import scan_components_from_source as _ref_sc
        return set(_ref_sc(source))
    return _scan_components_from_sx_fallback(source)


def components_needed(page_sx: str, env: dict[str, Any]) -> set[str]:
    """Compute the full set of component names needed for a page.

    Returns names with ~ prefix.
    """
    if _use_ref():
        from .ref.sx_ref import components_needed as _ref_cn
        return set(_ref_cn(page_sx, env))
    return _components_needed_fallback(page_sx, env)


def compute_all_io_refs(env: dict[str, Any], io_names: set[str]) -> None:
    """Compute and cache transitive IO refs for all Component entries in *env*."""
    if _use_ref():
        from .ref.sx_ref import compute_all_io_refs as _ref_cio
        _ref_cio(env, list(io_names))
        return
    _compute_all_io_refs_fallback(env, io_names)


def page_render_plan(page_sx: str, env: dict[str, Any], io_names: set[str] | None = None) -> dict[str, Any]:
    """Compute the render plan for a page.

    Returns dict with:
      - "components": {name: "server"|"client", ...}
      - "server": [names rendered server-side]
      - "client": [names rendered client-side]
      - "io-deps": [IO primitive names needed by server components]
    """
    if io_names is None:
        io_names = get_all_io_names()
    from .ref.sx_ref import page_render_plan as _ref_prp
    plan = _ref_prp(page_sx, env, list(io_names))
    return plan


def get_all_io_names() -> set[str]:
    """Build the complete set of IO primitive names from all boundary tiers.

    Includes: core IO (primitives_io.py handlers), plus all page helper names
    from every service boundary.
    """
    from .primitives_io import IO_PRIMITIVES
    from .boundary import declared_helpers

    names = set(IO_PRIMITIVES)
    for _svc, helper_names in declared_helpers().items():
        names.update(helper_names)
    return names