api/linter.py

"""
linter.py — Background Syntax Validator (S364)

Validazione sintattica leggera per file VFS. Mai blocca — sempre fire-and-forget.
Risultati salvati su Supabase (lint_ok, lint_errors, lint_at).

Supportato:
- Python: ast.parse() — zero deps, istantaneo
- JSON: json.loads()
- TypeScript/JavaScript: node --check + tsc --noEmit (P40-C: type errors visibili)
- TSX/JSX: S702 — _lint_tsx_jsx() structural checker (brace balance + JSX tag balance)
- Altro: skip silenzioso

Design: max _LINT_TIMEOUT=5s, silent failures, never raises

GAP-VFS-FIX: lint_and_store() ora accetta content_updated_at (BIGINT ms).
Se > 0, la UPDATE Supabase usa WHERE updated_at = content_updated_at.
Se 0 righe aggiornate → il file è stato aggiornato dopo l'inizio del lint →
risultato stale scartato silenziosamente. Previene lint_ok falso su contenuto
già superato da una scrittura concorrente.
"""
from __future__ import annotations
import ast as _ast
import asyncio
import json as _json
import os
import re
import tempfile

import logging
_logger = logging.getLogger("api.linter")

_LINT_TIMEOUT = 5.0  # secondi — mai blocca oltre questo


def _lint_python(content: str) -> dict:
    """ast.parse() — controllo sintattico Python senza dipendenze aggiuntive."""
    try:
        _ast.parse(content, mode='exec')
        return {'ok': True, 'errors': []}
    except SyntaxError as e:
        return {'ok': False, 'errors': [f"SyntaxError riga {e.lineno}: {e.msg}"]}
    except Exception as e:
        return {'ok': False, 'errors': [str(e)[:300]]}  # S588: 200→300


def _lint_json(content: str) -> dict:
    """Controllo sintattico JSON via json.loads()."""
    try:
        _json.loads(content)
        return {'ok': True, 'errors': []}
    except _json.JSONDecodeError as e:
        return {'ok': False, 'errors': [f"JSONDecodeError: {e.msg} (riga {e.lineno})"]}


async def _lint_tsc(fname: str) -> dict | None:
    """P40-C: TypeScript type-check via tsc --noEmit --skipLibCheck.

    Complementa node --check (solo sintassi V8) con type checking reale.
    Rileva: prop mancanti, interfacce incomplete, tipi incompatibili.
    Ritorna None se tsc non disponibile (fallback silenzioso a node --check).
    """
    try:
        proc = await asyncio.create_subprocess_exec(
            'tsc', '--noEmit', '--allowJs', '--skipLibCheck',
            '--strict', '--target', 'ESNext', '--moduleResolution', 'node',
            fname,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=8.0)
        if proc.returncode == 0:
            return {'ok': True, 'errors': [], 'checker': 'tsc'}
        # tsc scrive errori su stdout
        out_text = (stdout + stderr).decode('utf-8', errors='replace').strip()
        out_text = out_text.replace(fname, '<file>').replace('/tmp/', '')
        # Prendi le prime 3 righe di errore per non sovraccaricare
        lines = [l for l in out_text.splitlines() if 'error TS' in l][:3]
        err_msg = '\n'.join(lines) or out_text[:500]
        return {'ok': False, 'errors': [err_msg[:600]], 'checker': 'tsc'}
    except asyncio.TimeoutError:
        return {'ok': True, 'errors': [], 'skipped': 'tsc timeout', 'checker': 'tsc'}
    except FileNotFoundError:
        return None  # tsc non installato — usa solo node --check
    except Exception as e:
        _logger.debug("[linter] tsc silenced: %s", e)
        return None


async def _lint_node(content: str, ext: str) -> dict:
    """TypeScript/JavaScript syntax check via node --check + tsc --noEmit (P40-C)."""
    try:
        with tempfile.NamedTemporaryFile(suffix=f'.{ext}', mode='w',
                                          delete=False, dir='/tmp') as f:
            f.write(content)
            fname = f.name
        try:
            # Step 1: node --check (sintassi V8, veloce)
            proc = await asyncio.create_subprocess_exec(
                'node', '--check', fname,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=4.0)
            if proc.returncode != 0:
                err_text = stderr.decode('utf-8', errors='replace').strip()
                err_text = err_text.replace(fname, '<file>').replace('/tmp/', '')
                # S599: 300→500 — linter stderr può contenere path + messaggio lungo
                return {'ok': False, 'errors': [err_text[:500]], 'checker': 'node'}

            # Step 2 (P40-C): tsc --noEmit per .ts/.tsx — type errors reali
            if ext in ('ts', 'tsx'):
                tsc_result = await _lint_tsc(fname)
                if tsc_result is not None:
                    return tsc_result
                # GAP-LINT-FIX: tsc non installato → segnala esplicitamente
                # (prima: {'checker':'node'} senza flag → indistinguibile da successo)
                return {'ok': True, 'errors': [], 'checker': 'node', 'tsc_skipped': 'tsc_unavailable'}

            return {'ok': True, 'errors': [], 'checker': 'node'}
        finally:
            try:
                os.unlink(fname)
            except Exception as _exc:
                _logger.debug("[linter] silenced %s", type(_exc).__name__)  # noqa: BLE001
    except asyncio.TimeoutError:
        return {'ok': True, 'errors': [], 'skipped': 'node check timeout'}
    except FileNotFoundError:
        return {'ok': True, 'errors': [], 'skipped': 'node not found'}
    except Exception as e:
        return {'ok': True, 'errors': [], 'skipped': str(e)[:300]}  # S606: 200→300


# S702: HTML void elements — non richiedono tag di chiusura in JSX
_JSX_VOID_TAGS: frozenset[str] = frozenset({
    'br', 'hr', 'img', 'input', 'meta', 'link', 'area', 'base', 'col',
    'embed', 'param', 'source', 'track', 'wbr',
})


def _lint_tsx_jsx(content: str) -> dict:
    """
    S702: Structural syntax checker per TSX/JSX — zero dipendenze esterne.
    Controlla: bilanciamento brace/paren, bilanciamento tag JSX.
    Ritorna partial=True (non è un full parser — checker strutturale).
    """
    errors: list[str] = []

    # Strip commenti e stringhe per ridurre falsi positivi
    stripped = re.sub(r'//[^\n]*', '', content)
    stripped = re.sub(r'/\*[\s\S]*?\*/', '', stripped)
    stripped = re.sub(r'"(?:[^"\\]|\\.)*"', '""', stripped)
    stripped = re.sub(r"'(?:[^'\\]|\\.)*'", "''", stripped)
    stripped = re.sub(r'`(?:[^`\\]|\\.)*`', '``', stripped)

    # 1. Brace/paren balance
    brace_delta = stripped.count('{') - stripped.count('}')
    paren_delta = stripped.count('(') - stripped.count(')')
    if abs(brace_delta) > 2:
        errors.append(f"Parentesi graffe sbilanciate: {{}} (delta {brace_delta:+d})")
    if abs(paren_delta) > 2:
        errors.append(f"Parentesi tonde sbilanciate: () (delta {paren_delta:+d})")

    # 2. JSX tag balance — solo tag con nome (non tag vuoti HTML)
    tag_count: dict[str, int] = {}

    # Opening tags: <TagName (non self-closing, non closing, non comment)
    for m in re.finditer(r'<([A-Za-z][A-Za-z0-9.]*)(?:\s[^>]*)?>(?!\s*/)', content):
        tag = m.group(1)
        if tag.lower() not in _JSX_VOID_TAGS:
            tag_count[tag] = tag_count.get(tag, 0) + 1

    # Self-closing: <TagName ... />
    for m in re.finditer(r'<([A-Za-z][A-Za-z0-9.]*)[^>]*/>', content):
        tag = m.group(1)
        tag_count[tag] = tag_count.get(tag, 0) - 1

    # Closing: </TagName>
    for m in re.finditer(r'</([A-Za-z][A-Za-z0-9.]*)>', content):
        tag = m.group(1)
        tag_count[tag] = tag_count.get(tag, 0) - 1

    unbalanced = [
        (t, c) for t, c in tag_count.items()
        if abs(c) > 1 and t.lower() not in _JSX_VOID_TAGS
    ]
    for tag, delta in sorted(unbalanced, key=lambda x: -abs(x[1]))[:2]:
        errors.append(f"Tag JSX non bilanciato: <{tag}> (delta {delta:+d})")

    return {
        'ok':      len(errors) == 0,
        'errors':  errors,
        'partial': True,  # checker strutturale — non full parser
    }


async def lint_code(content: str, language: str, path: str = '') -> dict:
    """
    Entry point principale. Ritorna {ok, errors, language, path}.
    Mai rilancia eccezioni. Max _LINT_TIMEOUT secondi.
    """
    if not content or not content.strip():
        return {'ok': True, 'errors': [], 'language': language, 'path': path}

    lang = (language or '').lower().strip()
    ext  = path.rsplit('.', 1)[-1].lower() if '.' in path else ''

    try:
        if lang in ('python', 'py') or ext == 'py':
            result = _lint_python(content)
        elif lang in ('json',) or ext == 'json':
            result = _lint_json(content)
        elif lang in ('typescript', 'ts', 'javascript', 'js') or ext in ('ts', 'js'):
            result = await asyncio.wait_for(
                _lint_node(content, ext or ('ts' if 'typescript' in lang else 'js')),
                timeout=_LINT_TIMEOUT,
            )
        elif lang in ('tsx', 'jsx') or ext in ('tsx', 'jsx'):
            result = _lint_tsx_jsx(content)  # S702: structural checker (non più skip)
        else:
            result = {'ok': True, 'errors': [], 'skipped': f'no linter for {lang or ext or "unknown"}'}
    except asyncio.TimeoutError:
        result = {'ok': True, 'errors': [], 'skipped': 'lint timeout'}
    except Exception as e:
        result = {'ok': True, 'errors': [], 'skipped': str(e)[:300]}  # S606: 200→300

    result['language'] = language
    result['path']     = path
    return result


async def lint_and_store(
    file_id: str,
    content: str,
    language: str,
    path: str,
    conversation_id: str | None = None,
    content_updated_at: int = 0,
) -> dict:
    """
    Linta un file e salva il risultato su Supabase.

    P40-D: retry fino a 2 tentativi (0.5s sleep) invece di fire-and-forget puro.
    Copre il caso Railway cold-start dove le prime scritture cadono nel vuoto.

    GAP-VFS-FIX: content_updated_at (BIGINT ms, opzionale).
    Se > 0: la UPDATE usa WHERE id=? AND updated_at=content_updated_at.
    Se 0 righe aggiornate → il file ha ricevuto un'altra scrittura durante il lint
    → risultato scartato (stale lint prevention).
    Backward compatible: content_updated_at=0 (default) → comportamento originale.

    Ritorna il dict con il risultato del lint.
    """
    result = await lint_code(content, language, path)
    if file_id:
        import time as _time
        payload = {
            'lint_ok':     result.get('ok', True),
            'lint_errors': result.get('errors', []),
            'lint_at':     int(_time.time() * 1000),
        }
        _MAX_ATTEMPTS = 2
        for _attempt in range(_MAX_ATTEMPTS):
            try:
                from .state import sb
                q = sb().table('vfs_files').update(payload).eq('id', file_id)
                if content_updated_at > 0:
                    # GAP-VFS-FIX: confronto ottimistico su updated_at
                    q = q.eq('updated_at', content_updated_at)
                res = q.execute()
                # Controlla se la riga è stata effettivamente aggiornata
                if content_updated_at > 0 and not res.data:
                    _logger.info(
                        "[linter] GAP-VFS: lint result discarded for file_id=%s "
                        "(content updated_at=%d superseded by concurrent write)",
                        file_id, content_updated_at,
                    )
                    # Stale — non ritentare: il file ha una versione più recente
                    break
                break  # successo
            except Exception as _exc:
                if _attempt < _MAX_ATTEMPTS - 1:
                    await asyncio.sleep(0.5)  # P40-D: attendi breve prima del retry
                else:
                    _logger.debug("[linter] supabase store failed after %d attempts: %s",
                                  _MAX_ATTEMPTS, _exc)
                    # S364: silent — mai blocca VFS write
    return result