FamilyMealPlanner/scripts/check_cloudkit_schema.py

#!/usr/bin/env python3
"""Compares the SwiftData models against the deployed CloudKit schema.

Every stored property of a @Model becomes a CD_<name> field in CloudKit, and a
field missing from an environment simply doesn't sync: the app keeps working
and the data silently stays on the device that wrote it. That is how nine
fields drifted out of Production between 2.0 and 2.1.2 (issue #38).

The schema only picks up new fields when the app runs against the development
environment, and promoting them to Production is a manual step in CloudKit
Console. So this check has to run before shipping, not after.

    python3 scripts/check_cloudkit_schema.py [--environment production]

Exits non-zero when a stored property has no field in the schema.
"""

import argparse
import re
import subprocess
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
MODELS_DIR = REPO / "MealMood" / "Models"
CONTAINER = "iCloud.com.alexandrev.mealmood"

# Properties CloudKit does not mirror as a field of their own record type.
IGNORED = {
    # To-one relationships live on the child as CD_<name>; to-many ones have no
    # field at all (the inverse carries the link).
    "slots",
    "weekPlan",
}


def pass_show(entry: str) -> str:
    return subprocess.run(
        ["pass", "show", entry], capture_output=True, text=True, check=True
    ).stdout.splitlines()[0].strip()


def export_schema(environment: str) -> str:
    """Pulls the live schema. Needs the management token from `pass`."""
    token = pass_show("apple/mealmood/cloudkit-token")
    team = pass_show("apple/mealmood/developer-team-id")
    result = subprocess.run(
        [
            "xcrun", "cktool", "export-schema",
            "--token", token,
            "--team-id", team,
            "--container-id", CONTAINER,
            "--environment", environment,
        ],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise SystemExit(f"cktool export-schema failed:\n{result.stderr.strip()}")
    return result.stdout


def model_names() -> set:
    names = set()
    for path in MODELS_DIR.glob("*.swift"):
        names.update(re.findall(r"@Model\s+final class (\w+)", path.read_text()))
    return names


def stored_properties(source: str, models: set) -> list:
    """Stored `var`s of a @Model: computed ones end in `{`, and relationships to
    other models are not fields of this record type."""
    body = source.split("init(")[0]
    properties = []
    for match in re.finditer(r"^\s*(?:@Attribute\([^)]*\)\s*)?var (\w+)\s*:\s*([^\n=]+?)(\s*=|\s*\{|$)", body, re.M):
        name, type_name, tail = match.group(1), match.group(2).strip(), match.group(3)
        if tail.strip() == "{":
            continue  # computed
        if name in IGNORED:
            continue
        bare = type_name.strip("[]?").strip()
        if bare in models:
            continue  # relationship
        properties.append(name)
    return properties


def schema_fields(schema: str, record_type: str) -> set:
    block = re.search(rf"RECORD TYPE CD_{record_type} \((.*?)\);", schema, re.S)
    if not block:
        return set()
    # Binary attributes land as CD_<name>_ckAsset.
    return {
        field.removesuffix("_ckAsset")
        for field in re.findall(r"CD_(\w+)", block.group(1))
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--environment", default="production", choices=["production", "development"])
    args = parser.parse_args()

    schema = export_schema(args.environment)
    models = model_names()
    problems = []

    for path in sorted(MODELS_DIR.glob("*.swift")):
        source = path.read_text()
        for model in re.findall(r"@Model\s+final class (\w+)", source):
            fields = schema_fields(schema, model)
            if not fields:
                problems.append(f"{model}: record type CD_{model} missing from the schema")
                continue
            for prop in stored_properties(source, models):
                if prop not in fields:
                    problems.append(f"{model}.{prop} → CD_{prop} missing")

    if problems:
        print(f"CloudKit schema ({args.environment}) is behind the models:\n")
        for problem in problems:
            print(f"  ✗ {problem}")
        print(
            "\nThese fields will not sync. Run the app once against development so the\n"
            "schema picks them up (or import it with `cktool import-schema`), then deploy\n"
            "to Production from CloudKit Console → Schema → Deploy Schema Changes."
        )
        return 1

    print(f"✅ CloudKit schema ({args.environment}) matches the models.")
    return 0


if __name__ == "__main__":
    sys.exit(main())