nixos-servers/scripts/create-host/manipulators.py

"""Text manipulation for flake.nix and Terraform files."""

import re
from pathlib import Path

from models import HostConfig


def update_flake_nix(config: HostConfig, repo_root: Path, force: bool = False) -> None:
    """
    Add or update host entry in flake.nix nixosConfigurations.

    Args:
        config: Host configuration
        repo_root: Path to repository root
        force: If True, replace existing entry; if False, insert new entry
    """
    flake_path = repo_root / "flake.nix"
    content = flake_path.read_text()

    # Create new entry
    new_entry = f"""      {config.hostname} = nixpkgs.lib.nixosSystem {{
        inherit system;
        specialArgs = {{
          inherit inputs self sops-nix;
        }};
        modules = [
          (
            {{ config, pkgs, ... }}:
            {{
              nixpkgs.overlays = commonOverlays;
            }}
          )
          ./hosts/{config.hostname}
          sops-nix.nixosModules.sops
        ];
      }};
"""

    # Check if hostname already exists
    hostname_pattern = rf"^      {re.escape(config.hostname)} = nixpkgs\.lib\.nixosSystem"
    existing_match = re.search(hostname_pattern, content, re.MULTILINE)

    if existing_match and force:
        # Replace existing entry
        # Match the entire block from "hostname = " to "};"
        replace_pattern = rf"^      {re.escape(config.hostname)} = nixpkgs\.lib\.nixosSystem \{{.*?^      \}};\n"
        new_content, count = re.subn(replace_pattern, new_entry, content, flags=re.MULTILINE | re.DOTALL)

        if count == 0:
            raise ValueError(f"Could not find existing entry for {config.hostname} in flake.nix")
    else:
        # Insert new entry before closing brace of nixosConfigurations
        # Pattern: "      };\n      packages = forAllSystems"
        pattern = r"(      \};)\n(      packages = forAllSystems)"
        replacement = rf"{new_entry}\g<1>\n\g<2>"

        new_content, count = re.subn(pattern, replacement, content)

        if count == 0:
            raise ValueError(
                "Could not find insertion point in flake.nix. "
                "Looking for pattern: '      };\\n      packages = forAllSystems'"
            )

    flake_path.write_text(new_content)


def update_terraform_vms(config: HostConfig, repo_root: Path, force: bool = False) -> None:
    """
    Add or update VM entry in terraform/vms.tf locals.vms map.

    Args:
        config: Host configuration
        repo_root: Path to repository root
        force: If True, replace existing entry; if False, insert new entry
    """
    terraform_path = repo_root / "terraform" / "vms.tf"
    content = terraform_path.read_text()

    # Create new entry based on whether we have static IP or DHCP
    if config.is_static_ip:
        new_entry = f'''    "{config.hostname}" = {{
      ip        = "{config.ip}"
      cpu_cores = {config.cpu}
      memory    = {config.memory}
      disk_size = "{config.disk}"
    }}
'''
    else:
        new_entry = f'''    "{config.hostname}" = {{
      cpu_cores = {config.cpu}
      memory    = {config.memory}
      disk_size = "{config.disk}"
    }}
'''

    # Check if hostname already exists
    hostname_pattern = rf'^\s+"{re.escape(config.hostname)}" = \{{'
    existing_match = re.search(hostname_pattern, content, re.MULTILINE)

    if existing_match and force:
        # Replace existing entry
        # Match the entire block from "hostname" = { to }
        replace_pattern = rf'^\s+"{re.escape(config.hostname)}" = \{{.*?^\s+\}}\n'
        new_content, count = re.subn(replace_pattern, new_entry, content, flags=re.MULTILINE | re.DOTALL)

        if count == 0:
            raise ValueError(f"Could not find existing entry for {config.hostname} in terraform/vms.tf")
    else:
        # Insert new entry before closing brace
        # Pattern: "  }\n\n  # Compute VM configurations"
        pattern = r"(  \})\n\n(  # Compute VM configurations)"
        replacement = rf"{new_entry}\g<1>\n\n\g<2>"

        new_content, count = re.subn(pattern, replacement, content)

        if count == 0:
            raise ValueError(
                "Could not find insertion point in terraform/vms.tf. "
                "Looking for pattern: '  }\\n\\n  # Compute VM configurations'"
            )

    terraform_path.write_text(new_content)