Add privacy-safe telemetry lifecycle and outcome signals

scripts/check_telemetry_schema_parity.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+"""Verify the public telemetry payload and private receiver stay in lockstep."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import re
+import sys
+
+
+FIELD_RE = re.compile(
+    r'^\s*[A-Za-z0-9_]+\s+([A-Za-z0-9_]+)\s+`json:"([^",]+)(?:,[^"]*)?"`',
+    re.MULTILINE,
+)
+TYPESCRIPT_FIELD_RE = re.compile(
+    r"^\s*([a-z0-9_]+)\??:\s*(string|number|boolean);",
+    re.MULTILINE,
+)
+TYPESCRIPT_TO_GO = {"string": "string", "number": "int", "boolean": "bool"}
+LEGACY_RECEIVER_ONLY_FIELDS = {"license_tier", "api_tokens"}
+
+
+def struct_body(source: str, marker: str) -> str:
+    start = source.find(marker)
+    if start < 0:
+        raise ValueError(f"missing struct marker {marker!r}")
+    opening = source.find("{", start)
+    if opening < 0:
+        raise ValueError(f"missing opening brace after {marker!r}")
+
+    depth = 0
+    for index in range(opening, len(source)):
+        char = source[index]
+        if char == "{":
+            depth += 1
+        elif char == "}":
+            depth -= 1
+            if depth == 0:
+                return source[opening + 1 : index]
+    raise ValueError(f"unterminated struct after {marker!r}")
+
+
+def json_fields(source: str, marker: str) -> dict[str, str]:
+    fields: dict[str, str] = {}
+    for go_type, json_name in FIELD_RE.findall(struct_body(source, marker)):
+        if json_name in fields:
+            raise ValueError(f"duplicate JSON field {json_name!r} after {marker!r}")
+        fields[json_name] = go_type
+    if not fields:
+        raise ValueError(f"no JSON fields found after {marker!r}")
+    return fields
+
+
+def typescript_fields(source: str, marker: str) -> dict[str, str]:
+    fields: dict[str, str] = {}
+    for json_name, typescript_type in TYPESCRIPT_FIELD_RE.findall(struct_body(source, marker)):
+        if json_name in fields:
+            raise ValueError(f"duplicate TypeScript field {json_name!r} after {marker!r}")
+        fields[json_name] = TYPESCRIPT_TO_GO[typescript_type]
+    if not fields:
+        raise ValueError(f"no TypeScript fields found after {marker!r}")
+    return fields
+
+
+def parity_errors(
+    public_source: str,
+    receiver_source: str,
+    frontend_source: str | None = None,
+) -> list[str]:
+    public = json_fields(public_source, "type Ping struct")
+    receiver = json_fields(receiver_source, "var ping struct")
+    errors: list[str] = []
+
+    missing = sorted(set(public) - set(receiver))
+    if missing:
+        errors.append("receiver missing public fields: " + ", ".join(missing))
+
+    unexpected = sorted(set(receiver) - set(public) - LEGACY_RECEIVER_ONLY_FIELDS)
+    if unexpected:
+        errors.append("receiver-only fields outside the legacy allowlist: " + ", ".join(unexpected))
+
+    mismatched = sorted(
+        name
+        for name in set(public) & set(receiver)
+        if public[name] != receiver[name]
+    )
+    if mismatched:
+        errors.append(
+            "field type mismatches: "
+            + ", ".join(
+                f"{name} (public {public[name]}, receiver {receiver[name]})"
+                for name in mismatched
+            )
+        )
+
+    if public.get("schema_version") != "int":
+        errors.append("public payload must contain integer schema_version")
+
+    if frontend_source is not None:
+        frontend = typescript_fields(frontend_source, "export interface TelemetryPingPreview")
+        frontend_missing = sorted(set(public) - set(frontend))
+        if frontend_missing:
+            errors.append("frontend preview missing public fields: " + ", ".join(frontend_missing))
+
+        frontend_unexpected = sorted(set(frontend) - set(public))
+        if frontend_unexpected:
+            errors.append("frontend preview fields absent from public payload: " + ", ".join(frontend_unexpected))
+
+        frontend_mismatched = sorted(
+            name
+            for name in set(public) & set(frontend)
+            if public[name] != frontend[name]
+        )
+        if frontend_mismatched:
+            errors.append(
+                "frontend field type mismatches: "
+                + ", ".join(
+                    f"{name} (public {public[name]}, frontend {frontend[name]})"
+                    for name in frontend_mismatched
+                )
+            )
+    return errors
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--pulse-repo",
+        type=Path,
+        default=Path(__file__).resolve().parents[1],
+    )
+    parser.add_argument(
+        "--pulse-pro-repo",
+        type=Path,
+        default=Path(__file__).resolve().parents[2] / "pulse-pro",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv or sys.argv[1:])
+    public_path = args.pulse_repo / "internal" / "telemetry" / "telemetry.go"
+    frontend_path = args.pulse_repo / "frontend-modern" / "src" / "api" / "settings.ts"
+    receiver_path = args.pulse_pro_repo / "license-server" / "main.go"
+    if not public_path.is_file():
+        raise SystemExit(f"public telemetry source not found: {public_path}")
+    if not receiver_path.is_file():
+        raise SystemExit(f"private telemetry receiver source not found: {receiver_path}")
+    if not frontend_path.is_file():
+        raise SystemExit(f"frontend telemetry preview source not found: {frontend_path}")
+
+    errors = parity_errors(
+        public_path.read_text(),
+        receiver_path.read_text(),
+        frontend_path.read_text(),
+    )
+    if errors:
+        for error in errors:
+            print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+    print("telemetry schema parity: OK")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())