Synchronize public module directory publication

tools/release/govoplan_release/module_directory.py

@@ -20,6 +20,7 @@ def write_module_directory(
     output_root: Path,
     channel: str,
     public_base_url: str = DEFAULT_PUBLIC_BASE_URL,
+    prune: bool = False,
 ) -> tuple[Path, ...]:
     payloads = module_directory_payloads(
         catalog_payload=catalog_payload,
@@ -29,13 +30,62 @@ def write_module_directory(
     )
     written: list[Path] = []
     for relative_path, payload in payloads:
-        path = output_root / relative_path
-        path.parent.mkdir(parents=True, exist_ok=True)
+        path = _prepare_output_path(output_root=output_root, relative_path=relative_path)
         path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
         written.append(path)
+    if prune:
+        _prune_stale_module_directory_files(
+            output_root=output_root,
+            expected={relative_path for relative_path, _payload in payloads},
+        )
     return tuple(written)
 
 
+def _prepare_output_path(*, output_root: Path, relative_path: Path) -> Path:
+    if output_root.is_symlink():
+        raise ValueError("module-directory output root must not be a symlink")
+    output_root.mkdir(parents=True, exist_ok=True)
+    current = output_root
+    for part in relative_path.parent.parts:
+        current = current / part
+        if current.is_symlink():
+            raise ValueError("module-directory path must not contain symlinks")
+        if current.exists():
+            if not current.is_dir():
+                raise ValueError("module-directory parent path must be a directory")
+            continue
+        current.mkdir()
+    path = output_root / relative_path
+    if path.is_symlink() or (path.exists() and not path.is_file()):
+        raise ValueError("module-directory output path must be a regular file")
+    return path
+
+
+def _prune_stale_module_directory_files(
+    *,
+    output_root: Path,
+    expected: set[Path],
+) -> None:
+    module_root = output_root / "modules"
+    if module_root.is_symlink():
+        raise ValueError("module-directory root must not be a symlink")
+    if not module_root.exists():
+        return
+    for path in module_root.rglob("*.json"):
+        relative_path = path.relative_to(output_root)
+        if relative_path not in expected:
+            path.unlink()
+    for path in sorted(
+        (item for item in module_root.rglob("*") if item.is_dir()),
+        key=lambda item: len(item.parts),
+        reverse=True,
+    ):
+        try:
+            path.rmdir()
+        except OSError:
+            pass
+
+
 def module_directory_payloads(
     *,
     catalog_payload: dict[str, Any],