#!/usr/bin/env python3
"""abort_check.py: heuristic scan of JS/TS files for AbortController and AbortSignal hazards.

Usage: python3 abort_check.py [--strict] FILE_OR_DIR [...]
Reads files, writes nothing, makes no network calls. Standard library only.
Exit 0 = no errors (and no warnings with --strict). Exit 1 = findings. Exit 2 = bad input.
It reads text, not a syntax tree: a hit inside a string or comment is a false positive.
"""
import os
import re
import sys

EXT = (".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".mts", ".cts", ".vue", ".svelte")
SKIP = {"node_modules", ".git", "dist", "build", "coverage", ".next", ".nuxt"}


def strip(line):
    line = re.sub(r"//.*$", "", line)
    return re.sub(r"/\*.*?\*/", "", line)


def call_text(code, i):
    """Text of a call that starts on line i, up to its closing parenthesis (max 12 lines)."""
    text = " ".join(code[i:i + 12])
    start = text.find("(")
    depth = 0
    for k in range(start, len(text)):
        depth += text[k] == "("
        depth -= text[k] == ")"
        if depth == 0:
            return text[: k + 1]
    return text


def scan(path):
    out = []
    try:
        raw = open(path, encoding="utf-8", errors="replace").read()
    except OSError as e:
        print(f"{path}: cannot read ({e})", file=sys.stderr)
        sys.exit(2)
    code = [strip(l) for l in raw.splitlines()]
    whole = "\n".join(code)
    uses_abort = bool(re.search(r"AbortController|AbortSignal|\bsignal\b", whole))
    has_clear = "clearTimeout" in whole
    has_timeout_err = "TimeoutError" in whole
    has_remove = bool(re.search(r"removeEventListener\(\s*[\"']abort[\"']", whole))
    uses_timeout_signal = "AbortSignal.timeout" in whole
    first_catch = None
    in_effect = None
    effect_text = []
    for i, l in enumerate(code):
        n = i + 1
        if re.search(r"(^|[^\w.])fetch\(", l) and "signal" not in call_text(code, i):
            out.append((n, "warn", "fetch-no-signal", "fetch() with no signal: it cannot be cancelled or time out. Pass { signal }"))
        if re.search(r"setTimeout\(", l) and re.search(r"\.abort\(", " ".join(code[i:i + 3])) and not has_clear:
            out.append((n, "warn", "timeout-manual", "setTimeout that calls abort() and no clearTimeout in the file: the timer outlives a fast response. Use AbortSignal.timeout(ms), or clear it in finally"))
        if re.search(r"AbortError", l) and re.search(r"name\s*===?|\.name\b", l) and uses_timeout_signal and not has_timeout_err:
            out.append((n, "warn", "abort-name-only", "AbortSignal.timeout rejects with TimeoutError, not AbortError. Check both names"))
        m = re.search(r"addEventListener\(\s*[\"']abort[\"']", l)
        if m:
            if "once" not in call_text(code, i):
                out.append((n, "warn", "listener-no-once", "abort listener without { once: true }: it stays registered until the signal is collected"))
            elif not has_remove:
                out.append((n, "info", "listener-once-no-remove", "{ once: true } removes the listener only when abort fires. If the work ends first, remove it in finally"))
        if re.match(r"^(const|let|var)\s+\w+\s*=\s*new\s+AbortController\(", l):
            name = re.match(r"^(?:const|let|var)\s+(\w+)", l).group(1)
            if re.search(rf"\b{name}\.abort\(", whole):
                out.append((n, "warn", "controller-module-scope", f"module-level controller '{name}' is single use: after abort() every later call is rejected at once. Create one controller per operation"))
        if re.search(r"\buseEffect\(", l):
            in_effect, effect_text = n, []
        if in_effect:
            effect_text.append(l)
            if re.match(r"^\s*\}\s*,\s*\[", l) or (len(effect_text) > 60):
                block = " ".join(effect_text)
                if re.search(r"fetch\(|axios|\.get\(", block) and not re.search(r"abort|AbortController|ignore|cancel", block, re.I):
                    out.append((in_effect, "warn", "effect-fetch-no-abort", "useEffect starts a request with no abort in cleanup: a slow earlier response can overwrite a newer one. Create a controller per run and abort in the cleanup"))
                in_effect = None
        if first_catch is None and re.search(r"\bcatch\s*(\(|\{)", l) and uses_abort:
            first_catch = n
    if first_catch and not re.search(r"AbortError|TimeoutError|\.aborted|isAbort", whole):
        out.append((first_catch, "info", "catch-no-abort-check", "this file uses AbortSignal but no catch checks for AbortError or TimeoutError: a user cancel may be reported as a failure"))
    if re.search(r"\{\s*signal\s*[,}]|,\s*signal\s*\)|\(\s*signal\s*[,)]", whole) and re.search(r"\b(for|while)\b[^\n]*\{", whole) and "await" in whole \
            and not re.search(r"throwIfAborted|signal\??\.aborted|\.aborted", whole):
        idx = next((j + 1 for j, l in enumerate(code) if re.search(r"\bsignal\b", l)), 1)
        out.append((idx, "info", "signal-not-checked", "a function takes a signal and loops with await, but never calls signal.throwIfAborted(): a long loop ignores cancellation between steps"))
    return out


def walk(args):
    for a in args:
        if os.path.isdir(a):
            for root, dirs, files in os.walk(a):
                dirs[:] = [d for d in dirs if d not in SKIP]
                for f in sorted(files):
                    if f.endswith(EXT):
                        yield os.path.join(root, f)
        elif os.path.isfile(a):
            yield a
        else:
            print(f"{a}: not found", file=sys.stderr)
            sys.exit(2)


def main():
    argv = sys.argv[1:]
    strict = "--strict" in argv
    argv = [a for a in argv if a != "--strict"]
    if not argv:
        print(__doc__)
        sys.exit(2)
    rows, nfiles = [], 0
    for p in walk(argv):
        nfiles += 1
        rows += [(p,) + r for r in scan(p)]
    c = {"error": 0, "warn": 0, "info": 0}
    for r in rows:
        c[r[2]] += 1
    print(f"{c['error']} errors, {c['warn']} warnings, {c['info']} info, {nfiles} files")
    for p, n, lvl, code, msg in sorted(rows, key=lambda r: (r[0], r[1])):
        print(f"{p}:{n} {lvl:5} {code:24} {msg}")
    sys.exit(1 if c["error"] or (strict and c["warn"]) else 0)


if __name__ == "__main__":
    main()