1
#!/usr/bin/env python3
2
"""
3
Verify that every new compatibility flag with a default-on date
4
(via $compatEnableDate or $impliedByAfterDate) has matching documentation
5
in cloudflare/cloudflare-docs, either already merged to the production
6
branch or in an open PR with at least one approving review.
7
"""
8
 
9
import json
10
import os
11
import re
12
import subprocess
13
import sys
14
from pathlib import Path
15
 
16
CAPNP_PATH = "src/workerd/io/compatibility-date.capnp"
17
DOCS_REPO = "cloudflare/cloudflare-docs"
18
DOCS_BRANCH = "production"
19
DOCS_DIR = "src/content/compatibility-flags"
20
 
21
 
22
# ---------------------------------------------------------------------------
23
# capnp parsing
24
# ---------------------------------------------------------------------------
25
 
26
 
27
def parse_compat_flags(content: str) -> dict:
28
    """Return {enable_flag_name: info} for every flag that carries a
29
    $compatEnableDate or $impliedByAfterDate annotation."""
30
 
31
    result = {}
32
 
33
    # Split on field boundaries.  Each field starts with:
34
    #     fieldName @N :Type
35
    blocks = re.split(r"\n(?=\s*\w+\s+@\d+\s*)", content)
36
 
37
    for block in blocks:
38
        m = re.match(r"\s*(\w+)\s+@(\d+)\s*:\s*(\w+)", block)
39
        if not m:
40
            continue
41
 
42
        field_name = m.group(1)
43
 
44
        # Skip obsolete fields.
45
        if field_name.startswith("obsolete"):
46
            continue
47
 
48
        flag_m = re.search(r'\$compatEnableFlag\s*\(\s*"([^"]+)"', block)
49
        if not flag_m:
50
            continue
51
        enable_flag = flag_m.group(1)
52
 
53
        date_m = re.search(r'\$compatEnableDate\s*\(\s*"([^"]+)"', block)
54
        implied_m = re.search(r"\$impliedByAfterDate", block)
55
        is_experimental = "$experimental" in block
56
 
57
        if date_m or implied_m:
58
            result[enable_flag] = {
59
                "field_name": field_name,
60
                "enable_date": date_m.group(1) if date_m else None,
61
                "has_implied_by": implied_m is not None,
62
                "experimental": is_experimental,
63
            }
64
 
65
    return result
66
 
67
 
68
# ---------------------------------------------------------------------------
69
# git helpers
70
# ---------------------------------------------------------------------------
71
 
72
 
73
def get_base_content(base_sha: str) -> str:
74
    """Return the capnp file at *base_sha*, or '' if it doesn't exist."""
75
    r = subprocess.run(
76
        ["git", "show", f"{base_sha}:{CAPNP_PATH}"],
77
        capture_output=True,
78
        text=True,
79
    )
80
    return r.stdout if r.returncode == 0 else ""
81
 
82
 
83
def resolve_base_sha() -> str:
84
    """Determine the base SHA to diff against."""
85
    # 1. Explicit env var (set by the workflow for pull_request events).
86
    sha = os.environ.get("GITHUB_BASE_SHA")
87
    if sha:
88
        return sha
89
 
90
    # 2. Pull from the GitHub event payload.
91
    event_path = os.environ.get("GITHUB_EVENT_PATH")
92
    if event_path and Path(event_path).is_file():
93
        with Path(event_path).open() as f:
94
            event = json.load(f)
95
        sha = event.get("pull_request", {}).get("base", {}).get("sha") or event.get(
96
            "merge_group", {}
97
        ).get("base_sha")
98
        if sha:
99
            return sha
100
 
101
    # 3. Fallback - merge-base with origin/main.
102
    r = subprocess.run(
103
        ["git", "merge-base", "HEAD", "origin/main"],
104
        capture_output=True,
105
        text=True,
106
    )
107
    if r.returncode == 0:
108
        return r.stdout.strip()
109
 
110
    print("::error::Could not determine base SHA to diff against.")
111
    sys.exit(1)
112
 
113
 
114
# ---------------------------------------------------------------------------
115
# cloudflare-docs checks (via `gh` CLI)
116
# ---------------------------------------------------------------------------
117
 
118
 
119
def gh(*args, **kwargs) -> subprocess.CompletedProcess:
120
    """Run a `gh` command and return the CompletedProcess."""
121
    return subprocess.run(
122
        ["gh", *args],
123
        capture_output=True,
124
        text=True,
125
        **kwargs,
126
    )
127
 
128
 
129
def doc_exists_on_production(slug: str) -> bool:
130
    """Return True if *slug*.md exists on the production branch."""
131
    r = gh(
132
        "api",
133
        f"repos/{DOCS_REPO}/contents/{DOCS_DIR}/{slug}.md?ref={DOCS_BRANCH}",
134
        "-q",
135
        ".name",
136
    )
137
    return r.returncode == 0
138
 
139
 
140
def search_open_docs_prs(flag_name: str) -> list[dict]:
141
    """Search for open PRs in cloudflare-docs mentioning *flag_name*.
142
143
    Returns a (possibly empty) list of ``{"number": int, "title": str}``.
144
    """
145
    r = gh(
146
        "search",
147
        "prs",
148
        "--repo",
149
        DOCS_REPO,
150
        "--state",
151
        "open",
152
        flag_name,
153
        "--json",
154
        "number,title",
155
        "--limit",
156
        "10",
157
    )
158
    if r.returncode != 0 or not r.stdout.strip():
159
        return []
160
    try:
161
        return json.loads(r.stdout)
162
    except json.JSONDecodeError:
163
        return []
164
 
165
 
166
def pr_has_approval(pr_number: int) -> bool:
167
    """Return True if *pr_number* in cloudflare-docs has ≥1 approving review."""
168
    r = gh(
169
        "api",
170
        f"repos/{DOCS_REPO}/pulls/{pr_number}/reviews",
171
        "--jq",
172
        '[.[] | select(.state == "APPROVED")] | length',
173
    )
174
    if r.returncode != 0:
175
        return False
176
    try:
177
        return int(r.stdout.strip()) > 0
178
    except (ValueError, TypeError):
179
        return False
180
 
181
 
182
# ---------------------------------------------------------------------------
183
# main
184
# ---------------------------------------------------------------------------
185
 
186
 
187
def main() -> None:
188
    base_sha = resolve_base_sha()
189
 
190
    with Path(CAPNP_PATH).open() as f:
191
        head_content = f.read()
192
 
193
    base_content = get_base_content(base_sha)
194
 
195
    head_flags = parse_compat_flags(head_content)
196
    base_flags = parse_compat_flags(base_content)
197
 
198
    # Flags that are *new* or that *gained* a default-on date in this PR.
199
    new_flags: dict[str, dict] = {}
200
    for name, info in head_flags.items():
201
        if name not in base_flags:
202
            new_flags[name] = info
203
        else:
204
            old = base_flags[name]
205
            if info.get("enable_date") and not old.get("enable_date"):
206
                new_flags[name] = info
207
            elif info.get("has_implied_by") and not old.get("has_implied_by"):
208
                new_flags[name] = info
209
 
210
    if not new_flags:
211
        print("No new compatibility flags with default-on dates detected.")
212
        return
213
 
214
    print(f"Found {len(new_flags)} new flag(s) with default-on dates:\n")
215
    for name, info in sorted(new_flags.items()):
216
        date = info.get("enable_date") or "(implied-by-after-date)"
217
        print(f"  {name}  {date}")
218
    print()
219
 
220
    # ------------------------------------------------------------------
221
    # Check each flag for documentation.
222
    # ------------------------------------------------------------------
223
    undocumented: list[str] = []
224
    needs_approval: list[tuple[str, list[dict]]] = []
225
 
226
    for flag_name in sorted(new_flags):
227
        slug = flag_name.replace("_", "-")
228
 
229
        # 1. Already on the production branch?
230
        if doc_exists_on_production(slug):
231
            print(f"  ok   {flag_name}  (on production)")
232
            continue
233
 
234
        # 2. Open PR that mentions the flag?
235
        prs = search_open_docs_prs(flag_name)
236
        if not prs:
237
            # Also try the slug form (hyphens).
238
            prs = search_open_docs_prs(slug)
239
 
240
        if not prs:
241
            undocumented.append(flag_name)
242
            print(f"  FAIL {flag_name}  (no docs found)")
243
            continue
244
 
245
        # 3. Does any of those PRs have an approving review?
246
        approved = False
247
        for pr in prs:
248
            if pr_has_approval(pr["number"]):
249
                print(f"  ok   {flag_name}  (approved PR #{pr['number']})")
250
                approved = True
251
                break
252
 
253
        if not approved:
254
            needs_approval.append((flag_name, prs))
255
            nums = ", ".join(f"#{p['number']}" for p in prs)
256
            print(f"  WAIT {flag_name}  (PR {nums} needs approval)")
257
 
258
    # ------------------------------------------------------------------
259
    # Emit GitHub Actions error annotations.
260
    # ------------------------------------------------------------------
261
    print()
262
    errors: list[str] = []
263
 
264
    for flag_name in undocumented:
265
        slug = flag_name.replace("_", "-")
266
        msg = (
267
            f"Compatibility flag `{flag_name}` adds a default-on date "
268
            f"but has no documentation in {DOCS_REPO}.  "
269
            f"Please open a PR there adding "
270
            f"`{DOCS_DIR}/{slug}.md` "
271
            f"and get at least one approving review."
272
        )
273
        errors.append(msg)
274
        print(f"::error file={CAPNP_PATH}::{msg}")
275
 
276
    for flag_name, prs in needs_approval:
277
        nums = ", ".join(f"#{p['number']}" for p in prs)
278
        msg = (
279
            f"Compatibility flag `{flag_name}` has a docs PR ({nums}) "
280
            f"in {DOCS_REPO} but it still needs at least one approving review."
281
        )
282
        errors.append(msg)
283
        print(f"::error file={CAPNP_PATH}::{msg}")
284
 
285
    if errors:
286
        print()
287
        print(f"{len(errors)} flag(s) need documentation before this PR can merge.")
288
        print(
289
            f"\nSee https://github.com/{DOCS_REPO}/tree/{DOCS_BRANCH}/{DOCS_DIR} "
290
            "for examples."
291
        )
292
        sys.exit(1)
293
 
294
    print("All new flags with default-on dates are documented.")
295
 
296
 
297
if __name__ == "__main__":
298
    main()