register/scripts/migrate-nix.py

#!/usr/bin/env python3
"""
Migrate domains/*.json to domains/*.nix

Converts each JSON domain config into a .nix file matching the format
from docs/example.nix:

    { dns, ... }: {
      metadata = {
        description = "...";
        proxy = false;
        owner = {
          username = "...";
        };
      };
      records = with dns.lib.combinators; {
        CNAME = [ "example.com." ];
      };
    }

Usage:
    python3 scripts/migrate-nix.py [--dry-run] [--delete-json]

Options:
    --dry-run      Print generated .nix to stdout without writing files
    --delete-json  Delete the .json files after successful conversion
"""

import json
import sys
from pathlib import Path

DOMAINS_DIR = Path(__file__).resolve().parent.parent / "domains"


# --- Nix string helpers ---

def escape(s: str) -> str:
    """Escape a string for use inside Nix double quotes."""
    return s.replace("\\", "\\\\").replace('"', '\\"').replace("${", "\\${")


def fqdn(s: str) -> str:
    """Ensure a domain string ends with a trailing dot."""
    return s if s.endswith(".") else s + "."


# --- Block builders ---

def build_metadata(data: dict) -> list[str]:
    """Build the metadata = { ... }; block."""
    owner = data.get("owner", {})
    description = data.get("description")
    proxy = data.get("proxied", data.get("proxy"))

    lines = ["  metadata = {"]

    if description is not None:
        lines.append(f'    description = "{escape(description)}";')

    if proxy is not None:
        lines.append(f"    proxy = {'true' if proxy else 'false'};")

    owner_keys = ["username", "email", "discord", "repo"]
    owner_fields = [(k, owner[k]) for k in owner_keys if owner.get(k)]

    if owner_fields:
        lines.append("    owner = {")
        for key, val in owner_fields:
            lines.append(f'      {key} = "{escape(val)}";')
        lines.append("    };")

    lines.append("  };")
    return lines


def build_records(record: dict) -> list[str]:
    """Build the records = with dns.lib.combinators; { ... }; block."""
    entries = []

    # A records
    if "A" in record:
        entries.extend(string_list_record("A", as_list(record["A"])))

    # AAAA records
    if "AAAA" in record:
        entries.extend(string_list_record("AAAA", as_list(record["AAAA"])))

    # CNAME (also handles ALIAS → CNAME)
    cname = record.get("CNAME") or record.get("ALIAS")
    if cname is not None:
        val = cname[0] if isinstance(cname, list) else cname
        entries.append(f'    CNAME = [ "{fqdn(val)}" ];')

    # MX records
    if "MX" in record:
        entries.extend(build_mx(as_list(record["MX"])))

    # TXT records
    if "TXT" in record:
        escaped = [escape(v) for v in as_list(record["TXT"])]
        entries.extend(string_list_record("TXT", escaped))

    # NS records
    if "NS" in record:
        fqdns = [fqdn(v) for v in as_list(record["NS"])]
        entries.extend(string_list_record("NS", fqdns))

    # SRV records
    if "SRV" in record:
        entries.extend(build_srv(as_list(record["SRV"])))

    # CAA records
    if "CAA" in record:
        entries.extend(build_caa(as_list(record["CAA"])))

    if not entries:
        return ["  records = with dns.lib.combinators; {};"]

    lines = ["  records = with dns.lib.combinators; {"]
    lines.extend(entries)
    lines.append("  };")
    return lines


# --- Record type formatters ---

def as_list(value) -> list:
    """Wrap a scalar in a list if it isn't one already."""
    return value if isinstance(value, list) else [value]


def string_list_record(rtype: str, values: list[str]) -> list[str]:
    """Format a record type whose values are plain strings."""
    if len(values) == 1:
        return [f'    {rtype} = [ "{values[0]}" ];']

    lines = [f"    {rtype} = ["]
    for v in values:
        lines.append(f'      "{v}"')
    lines.append("    ];")
    return lines


def build_mx(values: list) -> list[str]:
    """Format MX records as attrsets with preference + exchange."""
    lines = ["    MX = ["]
    for i, v in enumerate(values):
        pref = (i + 1) * 10
        lines.append("      {")
        lines.append(f"        preference = {pref};")
        lines.append(f'        exchange = "{fqdn(v)}";')
        lines.append("      }")
    lines.append("    ];")
    return lines


def build_srv(values: list[dict]) -> list[str]:
    """Format SRV records."""
    lines = ["    SRV = ["]
    for srv in values:
        lines.append("      {")
        for key in ("service", "proto"):
            if key in srv:
                lines.append(f'        {key} = "{srv[key]}";')
        for key in ("priority", "weight", "port"):
            if key in srv:
                lines.append(f"        {key} = {srv[key]};")
        if "target" in srv:
            lines.append(f'        target = "{fqdn(srv["target"])}";')
        lines.append("      }")
    lines.append("    ];")
    return lines


def build_caa(values: list[dict]) -> list[str]:
    """Format CAA records."""
    lines = ["    CAA = ["]
    for caa in values:
        lines.append("      {")
        if "flags" in caa:
            lines.append(f"        flags = {caa['flags']};")
        if "tag" in caa:
            lines.append(f'        tag = "{caa["tag"]}";')
        if "value" in caa:
            lines.append(f'        value = "{escape(caa["value"])}";')
        lines.append("      }")
    lines.append("    ];")
    return lines


# --- Top-level conversion ---

def json_to_nix(data: dict) -> str:
    """Convert a parsed JSON domain config to a complete .nix file string."""
    lines = ["{ dns, ... }: {"]
    lines.extend(build_metadata(data))
    lines.extend(build_records(data.get("record", {})))
    lines.append("}")
    lines.append("")
    return "\n".join(lines)


# --- File operations ---

def migrate_file(path: Path, *, dry_run: bool, delete_json: bool) -> bool:
    """Migrate a single .json file. Returns True on success."""
    try:
        data = json.loads(path.read_text())
    except json.JSONDecodeError as e:
        print(f"  ERROR: {path.name}: {e}", file=sys.stderr)
        return False

    nix = json_to_nix(data)
    nix_path = path.with_suffix(".nix")

    if dry_run:
        print(f"--- {nix_path.name} ---")
        print(nix)
        return True

    nix_path.write_text(nix)
    print(f"  Created {nix_path.name}")

    if delete_json:
        path.unlink()
        print(f"  Deleted {path.name}")

    return True


def main():
    dry_run = "--dry-run" in sys.argv
    delete_json = "--delete-json" in sys.argv

    if not DOMAINS_DIR.exists():
        print(f"Error: {DOMAINS_DIR} not found", file=sys.stderr)
        sys.exit(1)

    files = sorted(DOMAINS_DIR.glob("*.json"))
    if not files:
        print("No .json files found in domains/")
        sys.exit(0)

    print(f"Found {len(files)} JSON file(s) to migrate")
    if dry_run:
        print("(dry run — no files will be written)\n")

    success = 0
    failed = 0

    for f in files:
        print(f"Migrating {f.name}...")
        if migrate_file(f, dry_run=dry_run, delete_json=delete_json):
            success += 1
        else:
            failed += 1

    print(f"\nDone: {success} succeeded, {failed} failed")
    if failed:
        sys.exit(1)


if __name__ == "__main__":
    main()