path: root/scripts/newos4-maintain.py

#!/usr/bin/env python3
"""newOS4 maintenance: drift, bundles, syntax, perms, data hygiene, live health.
Safe fixes only with --fix. Never touches nginx, services, sudo, or user content."""
import glob, json, os, re, subprocess, sys
from datetime import datetime, timezone

ROOT = '/var/www/webx11.duckdns.org/spaces/newos4'
PAGE = '/var/www/webx11.duckdns.org/spaces/agando/downloads/changelog.html'
WEB = 'https://webx11.duckdns.org/spaces/newos4'
FIX = '--fix' in sys.argv
DEEP = '--deep' in sys.argv
issues = []
def add(l, m): issues.append((l, m))
def run(cmd, **kw):
    kw.setdefault('timeout', 300)
    return subprocess.run(cmd, capture_output=True, text=True, **kw)
def sh(cmd): return run(cmd, shell=True).stdout.strip()

def check_versions():
    html = open(os.path.join(ROOT, 'main.html'), encoding='utf-8').read()
    loader = open(os.path.join(ROOT, 'modules/core/module-loader.js'), encoding='utf-8').read()
    m = re.search(r"const LATEST_VER\s*=\s*'([^']+)'", loader)
    ver = m.group(1) if m else '?'
    n = html.count(ver)
    add('PASS' if n == 3 else 'FAIL', f'version {ver} x{n} in main.html (expect 3)')

def syntax_failures():
    bad = []
    for p in glob.glob(os.path.join(ROOT, 'modules/core/*.js')) + glob.glob(os.path.join(ROOT, 'modules/ext/*.js')):
        if os.path.basename(p) in ('core-bundle.js', 'critical-ext-bundle.js'): continue
        if run(['node', '--check', p]).returncode: bad.append(os.path.basename(p))
    return bad

def check_bundles(bad_syntax=None):
    """Rebuild only when every module parses.

    rebuild-bundles.py concatenates blindly (no validation), so a module with a
    syntax error would be baked straight into the served core-bundle.js. Since
    --fix now runs daily, the rebuild is gated on check_syntax and the produced
    bundles are re-checked before the run is called clean.
    """
    paths = ('modules/core/core-bundle.js', 'modules/ext/critical-ext-bundle.js')
    before = {p: sh(f'md5sum {p}') for p in paths}
    if not FIX:
        add('PASS', 'bundle check skipped (use --fix)'); return
    if bad_syntax is None: bad_syntax = syntax_failures()
    if bad_syntax:
        add('FAIL', 'bundles rebuild BLOCKED: syntax fail in ' + ', '.join(bad_syntax[:5])); return
    r = run([sys.executable, 'scripts/rebuild-bundles.py'], cwd=ROOT)
    if r.returncode:
        add('FAIL', 'bundles FAILED'); return
    invalid = [p for p in paths if run(['node', '--check', os.path.join(ROOT, p)]).returncode]
    if invalid:
        add('FAIL', 'bundles rebuilt but INVALID: ' + ', '.join(invalid)); return
    after = {p: sh(f'md5sum {p}') for p in paths}
    changed = [p for p in paths if before[p] != after[p]]
    add('WARN' if changed else 'PASS',
        'bundles ' + ('rebuilt: ' + ', '.join(changed) if changed else 'clean'))

def check_syntax(bad=None):
    bad = syntax_failures() if bad is None else bad
    add('PASS' if not bad else 'FAIL', 'syntax ' + ('OK' if not bad else 'fail: ' + ', '.join(bad)))

def check_perms():
    targets = glob.glob(os.path.join(ROOT, 'course/*.json')) + glob.glob(os.path.join(ROOT, 'course/chapters/*.json')) \
        + [os.path.join(ROOT, 'modules/ext/course.js'), os.path.join(ROOT, 'course.html'), os.path.join(ROOT, 'api/course/event.php')]
    bad = []
    for p in targets:
        if not os.path.exists(p): continue
        if (os.stat(p).st_mode & 0o777) != 0o644 or sh(f'stat -c %G {p}') != 'www-data':
            if FIX:
                os.chmod(p, 0o644); sh(f'chgrp www-data {p}')
            else: bad.append(os.path.basename(p))
    add('PASS' if not bad else ('WARN' if FIX else 'FAIL'), 'perms ' + ('clean' if not bad else ('fixed' if FIX else 'bad: ' + ', '.join(bad[:5]))))

def check_index():
    idxp = os.path.join(ROOT, 'modules-index.json')
    atlas = json.load(open(os.path.join(ROOT, 'course/atlas.json')))
    idx = json.load(open(idxp))
    index_cmds = set(c for m in idx.get('modules', {}).values() for c in m.get('commands', []))
    missing = sorted(c for c, e in atlas.get('commands', {}).items() if not e.get('builtin') and c not in index_cmds)
    if not missing:
        add('PASS', f'modules-index covers {len(index_cmds)} commands'); return
    add('WARN', f'modules-index missing {len(missing)}: ' + ', '.join(missing[:5]))
    if not FIX: return
    for name, m in atlas['modules'].items():
        e = idx['modules'].get(name, {'name': name, 'file': m['file'], 'type': 'core' if m['tier'] == 'core-bundle' else 'ext',
                                      'class': '', 'auto_load': m['tier'] != 'ext-manual', 'description': ''})
        e.update({'commands': m['commands'], 'file': m['file'], 'tier': m['tier']})
        idx['modules'][name] = e
    idx['version'] = datetime.now(timezone.utc).strftime('%Y-%m-%d') + '-atlas'
    tmp = idxp + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as f: json.dump(idx, f, ensure_ascii=False, indent=1)
    os.replace(tmp, idxp); os.chmod(idxp, 0o644); sh(f'chgrp www-data {idxp}')
    add('PASS', 'modules-index regenerated')

def check_data():
    err = os.path.join(ROOT, 'data/debug/errors.json')
    if os.path.exists(err) and os.path.getsize(err) > 5 * 1024 * 1024:
        if FIX:
            arc = os.path.join(ROOT, 'data/debug/archive'); os.makedirs(arc, exist_ok=True)
            os.replace(err, os.path.join(arc, 'errors-' + datetime.now().strftime('%Y%m') + '.json'))
            open(err, 'w').write('[]'); os.chmod(err, 0o664); sh(f'chgrp www-data {err}')
            add('PASS', 'errors.json rotated')
        else: add('WARN', 'errors.json >5MB')
    else: add('PASS', 'debug data OK')

def check_live():
    for url in (f'{WEB}/main.html', f'{WEB}/course.html', f'{WEB}/course/course.json', f'{WEB}/modules/core/core-bundle.js'):
        code = sh(f"curl -s -o /dev/null -w '%{{http_code}}' '{url}'")
        add('PASS' if code == '200' else 'FAIL', f'HTTP {code} {url.replace(WEB, "")}')

def check_atlas():
    p = os.path.join(ROOT, 'course/atlas.json')
    before = (json.load(open(p)).get('coverage', {}).get('commands') if os.path.exists(p) else None)
    r = run([sys.executable, 'scripts/gen-course-atlas.py'], cwd=ROOT)
    after = (json.load(open(p)).get('coverage', {}).get('commands') if os.path.exists(p) else None)
    add('PASS' if r.returncode == 0 else 'FAIL',
        f'atlas {after} commands' + (' (refreshed from ' + str(before) + ')' if before != after else ''))

def check_course():
    r = run([sys.executable, 'scripts/course-validate.py'], cwd=ROOT)
    add('PASS' if r.returncode == 0 else 'FAIL', 'course ' + ('valid' if r.returncode == 0 else 'INVALID: ' + r.stdout[-160:]))

def check_build_tags():
    """--deep only: every build tag must be reachable from HEAD and must name the version its
    commit actually carries.

    The changelog trusts tags (one `for-each-ref` pass instead of a `git describe` per commit).
    That trust is cheap to audit but catastrophic to lose silently: a tag moved onto another
    commit reassigns that commit to a different build, and a rewritten history orphans the whole
    set, with no error anywhere. 67 `git show` calls, so it runs under --deep only.
    """
    if not DEEP:
        return
    try:
        out = sh('git -C /var/www/webx11.duckdns.org tag --merged HEAD')
    except Exception as exc:
        add('FAIL', 'build tags: git failed (%s)' % exc)
        return
    tags = [t for t in out.split() if re.match(r'^v?\d+(\.\d+)+$', t)]
    if not tags:
        add('WARN', 'build tags: none reachable from HEAD')
        return
    unreachable = sh('git -C /var/www/webx11.duckdns.org tag').split()
    unreachable = [t for t in unreachable if re.match(r'^v?\d+(\.\d+)+$', t) and t not in tags]
    bad = []
    for tag in tags:
        try:
            src = sh("git -C /var/www/webx11.duckdns.org show '%s:spaces/newos4/modules/core/module-loader.js'" % tag)
        except Exception:
            bad.append(tag + ' (unreadable)')
            continue
        m = re.search(r"LATEST_VER\s*=\s*'([^']+)'", src or '')
        if not m or m.group(1).strip() != tag:
            bad.append('%s (tree says %s)' % (tag, m.group(1).strip() if m else 'nothing'))
    if unreachable:
        add('FAIL', 'build tags orphaned from HEAD: ' + ', '.join(unreachable[:4]))
    if bad:
        add('FAIL', 'build tags disagree with their commit: ' + '; '.join(bad[:4]))
    if not unreachable and not bad:
        add('PASS', 'build tags: %d reachable, all match their commit' % len(tags))


def check_changelog():
    """The changelog page must describe the running build and must be freshly generated.

    Three separate failures, three separate signals:
      - version mismatch  -> the page describes an older build (regenerate);
      - json/feed drift   -> the page is current but a sibling artifact is not (regenerate);
      - age > 30h         -> the daily job is not running (only after a regeneration attempt).

    Under --fix we always regenerate first. That is deliberate: the generator costs 0.84 s, and a
    page older than 30h is the NORMAL outcome of a quiet week with no version bumps. The first cut
    only regenerated on a version mismatch, so a quiet fortnight turned into a permanent FAIL that
    --fix could not clear - a red line that means "nothing happened", which trains the reader to
    ignore red.
    """
    script = os.path.join(ROOT, 'scripts/build-changelog.py')
    if not os.path.exists(script):
        add('FAIL', 'changelog: scripts/build-changelog.py missing')
        return
    if not os.path.exists(PAGE):
        add('FAIL', 'changelog: page missing - restore with '
                    '`git checkout -- spaces/agando/downloads/changelog.html`')
        return

    if FIX:
        r = run([sys.executable, script], cwd=ROOT)
        if r.returncode != 0:
            add('FAIL', 'changelog regen failed: ' + (r.stderr.strip()[:150] or '?'))
            return

    r = run([sys.executable, script, '--check'], cwd=ROOT)
    if r.returncode != 0:
        stale = (r.stdout or r.stderr).strip().splitlines()
        add('FAIL', 'changelog ' + (stale[-1][:150] if stale else 'stale'))
        return
    add('PASS', 'changelog: ' + (r.stdout.strip() or 'current'))

    # The page can be current while its siblings lag (a crash between writes). Compare, don't assume.
    try:
        js = json.load(open(os.path.join(os.path.dirname(PAGE), 'changelog.json'), encoding='utf-8'))
    except Exception as exc:
        add('FAIL', 'changelog.json unreadable: %s' % exc)
        return
    if js.get('currentVersion') != re.search(r"LATEST_VER\s*=\s*'([^']+)'",
            open(os.path.join(ROOT, 'modules/core/module-loader.js'), encoding='utf-8').read()).group(1):
        add('FAIL', 'changelog.json is on %s, build is newer - feed drifted'
            % js.get('currentVersion'))
        return

    import datetime as _dt
    stamp = js.get('checked') or js.get('generated')
    if not stamp:
        add('FAIL', 'changelog: no checked/generated timestamp in changelog.json')
        return
    try:
        gen = _dt.datetime.strptime(stamp, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=_dt.timezone.utc)
    except ValueError:
        add('FAIL', 'changelog: unreadable timestamp ' + str(stamp))
        return
    age_h = (_dt.datetime.now(_dt.timezone.utc) - gen).total_seconds() / 3600.0
    if age_h > 30:
        add('FAIL', 'changelog: generated %.0fh ago (>30h) - the daily job is not running '
                    '(run maintain --fix to refresh)' % age_h)
    else:
        add('PASS', 'changelog age %.1fh' % age_h)


def check_help_coverage():
    """Every command the code registers must appear in help.html (2026-10-22).

    help.html had a 74-command backlog because nothing compared the two. This
    check makes the gap visible on every maintenance run.
    """
    cmds, term = set(), open(os.path.join(ROOT, 'modules/core/terminal.js'), encoding='utf-8').read()
    for m in re.finditer(r'this\._baseWhitelist = \[(.*?)\];', term, re.S):
        cmds.update(re.findall(r"'([^']+)'", m.group(1)))
    for m in re.finditer(r'const internals = \[(.*?)\];', term, re.S):
        cmds.update(re.findall(r"'([^']+)'", m.group(1)))
    cmds.update(re.findall(r"cmd === ['\"]([a-z0-9_-]+)['\"]", term))
    for p in glob.glob(os.path.join(ROOT, 'modules/**/*.js'), recursive=True):
        if 'bundle' in p or '/test/' in p or '.bkp' in p:
            continue
        src = open(p, encoding='utf-8', errors='ignore').read()
        cmds.update(re.findall(r"commands\[['\"]([a-z0-9_-]+)['\"]\]\s*=", src))
    doc = open(os.path.join(ROOT, 'help.html'), encoding='utf-8').read()
    documented = set()
    for tok in re.findall(r'<code>([^<]+)</code>', doc):
        first = tok.replace('&lt;', ' ').replace('&gt;', ' ').split()
        if first and re.fullmatch(r'[a-z][a-z0-9_-]{0,40}', first[0].lstrip('./')):
            documented.add(first[0].lstrip('./'))
    missing = sorted(cmds - documented)
    add('PASS' if not missing else 'WARN',
        f'help coverage {len(cmds) - len(missing)}/{len(cmds)} commands'
        + ('' if not missing else f' — undocumented: {", ".join(missing[:8])}'))

def check_regression_tests():
    """Every node-runnable test under modules/test/ must pass (2026-10-27).

    The v4.6.102 checkpoint regression was written but no job executed it, so drift could ship
    silently. Two header conventions exist: '// ... run: node' and '// node modules/test/x.js'.
    test-suite.js starts with a class (browser-only) and is correctly skipped.
    """
    tests, failed = [], []
    for p in sorted(glob.glob(os.path.join(ROOT, 'modules/test/test-*.js'))):
        head = ''.join(open(p, encoding='utf-8', errors='ignore').readlines()[:3])
        if 'run: node' not in head and not head.startswith('// node modules/test/'):
            continue
        tests.append(p)
        try:
            r = subprocess.run(['node', p], cwd=ROOT, capture_output=True, text=True, timeout=120)
        except subprocess.TimeoutExpired:
            failed.append(os.path.basename(p) + ': timeout')
            continue
        if r.returncode != 0:
            lines = (r.stdout or r.stderr).strip().splitlines() or ['no output']
            failed.append(os.path.basename(p) + ': ' + lines[-1][:120])
    add('PASS' if not failed else 'FAIL',
        'node tests %d/%d pass' % (len(tests) - len(failed), len(tests))
        + ('' if not failed else ' — ' + '; '.join(failed[:3])))

def check_links():
    if not DEEP: return
    broken = []
    for page in ('main.html', 'course.html', 'help.html', 'modules.html', 'guide.html'):
        p = os.path.join(ROOT, page)
        if not os.path.exists(p): continue
        for href in re.findall(r'(?:href|src)="([^"#]+)"', open(p, encoding='utf-8').read()):
            if any(x in href for x in ("'", '+', '{', '}')): continue
            if href.startswith(('http', '//', 'mailto:', 'javascript:', 'data:', '/api/')): continue
            target = os.path.normpath(os.path.join(ROOT, href.split('?')[0]))
            if not os.path.exists(target): broken.append(f'{page}:{href}')
    add('PASS' if not broken else 'WARN', 'links ' + ('OK' if not broken else f'{len(broken)} broken (first: {broken[0]})'))

def main():
    os.chdir(ROOT)
    bad = syntax_failures()          # computed once, gates the bundle rebuild
    check_versions(); check_syntax(bad); check_regression_tests(); check_bundles(bad); check_perms(); check_index(); check_data(); check_live(); check_atlas(); check_course(); check_help_coverage(); check_links(); check_changelog(); check_build_tags()
    fails = [m for l, m in issues if l == 'FAIL']
    print('newOS4 maintenance — ' + datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%MZ') + (' [fix]' if FIX else '') + (' [deep]' if DEEP else ''))
    for l, m in issues: print(f'{l}: {m}')
    print('maintenance result: ' + ('FAIL' if fails else 'OK'))
    print('needs OK: ' + ('none' if not fails else '; '.join(fails)[:400]))
    return 1 if fails else 0

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