"""One route identity for catalog install and the W4P bundler. The bundler erases the name inside ``[brackets]``. ``[form_key]`` and ``[formKey]`` are the same route, and so are ``[entity]`` and ``[entity_key]``. Install must remove the extra file for every capability, not for forms only. """ from __future__ import annotations import re from pathlib import Path from src.domain.codegen.local_backup import move_to_bak _API_ROOTS = ("functions/api/", "server/api/") _METHOD_SUFFIX = { "Get": "GET", "Post": "POST", "Put": "PUT", "Patch": "PATCH", "Delete": "DELETE", "Head": "HEAD", "Options": "OPTIONS", } _NAMED_HANDLER = re.compile(r"export\s+(?:async\s+)?function\s+onRequest([A-Za-z]*)\b") _ALIASED_HANDLER = re.compile(r"export\s*\{([^}]+)\}") _SOURCE_SUFFIX = {".js", ".mjs", ".ts", ".tsx", ".cjs"} def application_pattern(rel: str) -> str | None: """Bundler pattern for one Pages or Nitro API file. Parameter names are erased.""" normalized = rel.replace("\\", "/") root = next((prefix for prefix in _API_ROOTS if normalized.startswith(prefix)), None) if root is None: return None rest = normalized[len(root):] rest = re.sub(r"\.(?:mjs|cjs|jsx|tsx|js|ts)$", "", rest) rest = re.sub( r"\.(?:get|post|put|patch|delete|head|options)$", "", rest, flags=re.IGNORECASE, ) if rest.endswith("/index"): rest = rest[: -len("/index")] segments = ["api"] for part in [item for item in rest.split("/") if item]: if part.startswith("[[") and part.endswith("]]"): segments.append("**") elif part.startswith("[...") and part.endswith("]"): segments.append("***") elif part.startswith("[") and part.endswith("]"): if not part[1:-1]: return None segments.append("*") else: segments.append(part) return f"{root}|{'/'.join(segments)}" def handler_methods(source: str) -> set[str]: """Methods a Pages Function exports. Empty means the file is not a handler.""" methods: set[str] = set() for match in _NAMED_HANDLER.finditer(source): suffix = match.group(1) if not suffix: methods.add("*") elif suffix in _METHOD_SUFFIX: methods.add(_METHOD_SUFFIX[suffix]) for match in _ALIASED_HANDLER.finditer(source): for part in match.group(1).split(","): item = part.strip() if not item: continue pieces = re.split(r"\s+as\s+", item) name = (pieces[1] if len(pieces) > 1 else pieces[0]).strip() if name == "onRequest": methods.add("*") elif name.startswith("onRequest"): suffix = name[len("onRequest"):] if suffix in _METHOD_SUFFIX: methods.add(_METHOD_SUFFIX[suffix]) return methods def _methods_overlap(left: set[str], right: set[str]) -> bool: if not left or not right or "*" in left or "*" in right: return True return bool(left & right) def _prefer(rel: str) -> tuple: snake_params = len(re.findall(r"\[[^\]]*_", rel)) return (snake_params, len(rel), rel) def route_methods_for_path(rel: str) -> set[str]: """Method claimed by a filename. Unsuffixed Pages handlers claim every method.""" normalized = rel.replace("\\", "/") stem = re.sub(r"\.(?:mjs|cjs|jsx|tsx|js|ts)$", "", normalized) match = re.search(r"\.(get|post|put|patch|delete|head|options)$", stem, re.IGNORECASE) if match is None: return {"*"} return {match.group(1).upper()} def catalog_route_collisions(paths: set[str]) -> list[tuple[str, str]]: """Pairs of catalog files the bundler would reject as one route.""" grouped: dict[str, list[str]] = {} for rel in paths: pattern = application_pattern(rel) if pattern is not None: grouped.setdefault(pattern, []).append(rel) collisions: list[tuple[str, str]] = [] for group in grouped.values(): ordered = sorted(group) for index, left in enumerate(ordered): left_methods = route_methods_for_path(left) for right in ordered[index + 1:]: if _methods_overlap(left_methods, route_methods_for_path(right)): collisions.append((left, right)) return collisions def retire_colliding_application_routes(repo_path: str, keep: set[str]) -> list[str]: """Move extra files that claim a route the catalog, or another file, already owns. ``keep`` is the set of paths this install will leave in place. A file on disk that normalizes to the same route is removed even when the kept file is not written yet. Two kept paths for one route are a catalog error. """ owned = [ pair for pair in catalog_route_collisions(keep) if pair[0].startswith("functions/api/") ] if owned: left, right = owned[0] raise ValueError(f"catalog_route_collision:{left} {right}") root = Path(repo_path) discovered: list[tuple[str, str, set[str]]] = [] for prefix in _API_ROOTS: directory = root / prefix.rstrip("/") if not directory.is_dir(): continue for path in directory.rglob("*"): if not path.is_file() or path.suffix not in _SOURCE_SUFFIX: continue rel = path.relative_to(root).as_posix() pattern = application_pattern(rel) if pattern is None: continue try: source = path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError): source = "" methods = handler_methods(source) or {"*"} discovered.append((rel, pattern, methods)) by_pattern: dict[str, list[tuple[str, set[str]]]] = {} for rel, pattern, methods in discovered: by_pattern.setdefault(pattern, []).append((rel, methods)) for rel in keep: pattern = application_pattern(rel) if pattern is None: continue group = by_pattern.setdefault(pattern, []) if any(existing == rel for existing, _methods in group): continue group.append((rel, route_methods_for_path(rel))) retired: list[str] = [] for group in by_pattern.values(): winners = {rel for rel, _methods in group if rel in keep} if not winners: on_disk = [rel for rel, _methods in group if (root / rel).is_file()] if len(on_disk) < 2: continue winners = {min(on_disk, key=_prefer)} for rel, methods in group: if rel in winners or not (root / rel).is_file(): continue winner_methods = [ methods for winner, methods in group if winner in winners ] if not any(_methods_overlap(methods, winner) for winner in winner_methods): continue retired.append(move_to_bak(root, rel)) return retired