```python
#!/usr/bin/env python3
"""Node.js memory leak investigation helper.

This script helps operators collect repeatable memory samples from a Node.js
process and summarize whether memory growth looks suspicious.

What it does:
- Reads a CSV file of memory samples or a JSON file with the same fields.
- Validates the input.
- Calculates basic trend signals for RSS, heap used, and heap total.
- Flags patterns that often suggest a leak or external/native memory growth.
- Prints a vendor-neutral investigation summary.

Expected sample fields:
- timestamp: ISO-8601 string or any sortable text value
- rss_mb: resident set size in MB
- heap_used_mb: V8 heap used in MB
- heap_total_mb: V8 heap total in MB
- external_mb: optional external memory in MB
- gc_pause_ms: optional garbage collection pause time in ms

Example CSV:

timestamp,rss_mb,heap_used_mb,heap_total_mb,external_mb,gc_pause_ms
2026-08-07T10:00:00Z,180,92,120,18,4
2026-08-07T10:05:00Z,190,97,128,18,5
2026-08-07T10:10:00Z,205,110,136,19,7

This script is intentionally read-only and does not connect to any service.
"""

from __future__ import annotations

import argparse
import csv
import json
import statistics
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, List, Optional


@dataclass
class Sample:
    timestamp: str
    rss_mb: float
    heap_used_mb: float
    heap_total_mb: float
    external_mb: Optional[float] = None
    gc_pause_ms: Optional[float] = None


def parse_float(value: str, field_name: str) -> float:
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"Invalid numeric value for {field_name}: {value!r}") from exc


def load_csv(path: Path) -> List[Sample]:
    samples: List[Sample] = []
    with path.open("r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        required = {"timestamp", "rss_mb", "heap_used_mb", "heap_total_mb"}
        missing = required - set(reader.fieldnames or [])
        if missing:
            raise ValueError(f"CSV is missing required columns: {', '.join(sorted(missing))}")

        for idx, row in enumerate(reader, start=2):
            timestamp = (row.get("timestamp") or "").strip()
            if not timestamp:
                raise ValueError(f"Row {idx}: timestamp is required")

            samples.append(
                Sample(
                    timestamp=timestamp,
                    rss_mb=parse_float(row.get("rss_mb", ""), "rss_mb"),
                    heap_used_mb=parse_float(row.get("heap_used_mb", ""), "heap_used_mb"),
                    heap_total_mb=parse_float(row.get("heap_total_mb", ""), "heap_total_mb"),
                    external_mb=parse_optional_float(row.get("external_mb")),
                    gc_pause_ms=parse_optional_float(row.get("gc_pause_ms")),
                )
            )
    return samples


def load_json(path: Path) -> List[Sample]:
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, list):
        raise ValueError("JSON input must be a list of sample objects")

    samples: List[Sample] = []
    for idx, item in enumerate(data, start=1):
        if not isinstance(item, dict):
            raise ValueError(f"Item {idx} must be an object")
        timestamp = str(item.get("timestamp", "")).strip()
        if not timestamp:
            raise ValueError(f"Item {idx}: timestamp is required")
        samples.append(
            Sample(
                timestamp=timestamp,
                rss_mb=parse_float(item["rss_mb"], "rss_mb") if "rss_mb" in item else fail(idx, "rss_mb"),
                heap_used_mb=parse_float(item["heap_used_mb"], "heap_used_mb") if "heap_used_mb" in item else fail(idx, "heap_used_mb"),
                heap_total_mb=parse_float(item["heap_total_mb"], "heap_total_mb") if "heap_total_mb" in item else fail(idx, "heap_total_mb"),
                external_mb=parse_optional_float(item.get("external_mb")),
                gc_pause_ms=parse_optional_float(item.get("gc_pause_ms")),
            )
        )
    return samples


def fail(idx: int, field: str):
    raise ValueError(f"Item {idx}: {field} is required")


def parse_optional_float(value) -> Optional[float]:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"Invalid optional numeric value: {value!r}") from exc


def linear_slope(values: List[float]) -> float:
    if len(values) < 2:
        return 0.0
    xs = list(range(len(values)))
    x_mean = statistics.mean(xs)
    y_mean = statistics.mean(values)
    num = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, values))
    den = sum((x - x_mean) ** 2 for x in xs)
    return num / den if den else 0.0


def summarize(samples: List[Sample]) -> str:
    rss = [s.rss_mb for s in samples]
    heap_used = [s.heap_used_mb for s in samples]
    heap_total = [s.heap_total_mb for s in samples]
    external = [s.external_mb for s in samples if s.external_mb is not None]
    gc = [s.gc_pause_ms for s in samples if s.gc_pause_ms is not None]

    rss_slope = linear_slope(rss)
    heap_used_slope = linear_slope(heap_used)
    heap_total_slope = linear_slope(heap_total)
    external_slope = linear_slope(external) if len(external) >= 2 else 0.0

    rss_growth = rss[-1] - rss[0]
    heap_growth = heap_used[-1] - heap_used[0]

    lines = []
    lines.append("Node.js Memory Investigation Summary")
    lines.append("=" * 40)
    lines.append(f"Samples collected: {len(samples)}")
    lines.append(f"RSS start/end: {rss[0]:.1f} MB -> {rss[-1]:.1f} MB (delta {rss_growth:+.1f} MB)")
    lines.append(f"Heap used start/end: {heap_used[0]:.1f} MB -> {heap_used[-1]:.1f} MB (delta {heap_growth:+.1f} MB)")
    lines.append(f"Heap total start/end: {heap_total[0]:.1f} MB -> {heap_total[-1]:.1f} MB")

    if external:
        lines.append(f"External memory start/end: {external[0]:.1f} MB -> {external[-1]:.1f} MB")

    if gc:
        lines.append(f"Average GC pause: {statistics.mean(gc):.1f} ms")

    lines.append("")
    lines.append("Trend signals:")
    lines.append(f"- RSS slope: {rss_slope:.3f} MB/sample")
    lines.append(f"- Heap used slope: {heap_used_slope:.3f} MB/sample")
    lines.append(f"- Heap total slope: {heap_total_slope:.3f} MB/sample")
    if external:
        lines.append(f"- External slope: {external_slope:.3f} MB/sample")

    suspicion = []
    if rss_slope > 0.5 and heap_used_slope <= 0.2:
        suspicion.append("RSS is climbing faster than heap used; inspect Buffers, native allocations, and non-heap memory.")
    if heap_used_slope > 0.2:
        suspicion.append("Heap used is steadily increasing; inspect retained JavaScript objects, maps, listeners, timers, and closures.")
    if gc and statistics.mean(gc) > 10:
        suspicion.append("GC pauses are elevated; verify whether the process is repeatedly reclaiming memory without returning to baseline.")
    if not suspicion:
        suspicion.append("No strong leak signal detected from these samples alone; verify under longer steady-state load and compare after GC.")

    lines.append("")
    lines.append("Assessment:")
    for item in suspicion:
        lines.append(f"- {item}")

    lines.append("")
    lines.append("Suggested next steps:")
    lines.append("1. Repeat sampling under stable traffic and capture post-GC baselines.")
    lines.append("2. Take heap snapshots before and after the workload window.")
    lines.append("3. Review retention points: listeners, timers, maps, closures, request-scoped objects, and caches.")
    lines.append("4. If RSS grows faster than heap, inspect Buffers, streams, native modules, and external memory use.")
    lines.append("5. Validate the fix under repeatable load before rollout.")

    return "\n".join(lines)


def read_samples(path: Path) -> List[Sample]:
    suffix = path.suffix.lower()
    if suffix == ".csv":
        return load_csv(path)
    if suffix == ".json":
        return load_json(path)
    raise ValueError("Input file must be .csv or .json")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Summarize Node.js memory samples for leak investigation."
    )
    parser.add_argument("input", help="Path to a CSV or JSON file with memory samples")
    parser.add_argument(
        "--output",
        help="Optional path to write the report. If omitted, prints to stdout.",
        default=None,
    )
    args = parser.parse_args()

    input_path = Path(args.input)
    if not input_path.exists():
        print(f"Input file not found: {input_path}", file=sys.stderr)
        return 1

    try:
        samples = read_samples(input_path)
        if len(samples) < 2:
            raise ValueError("At least two samples are required for trend analysis")
        report = summarize(samples)
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    if args.output:
        output_path = Path(args.output)
        output_path.write_text(report + "\n", encoding="utf-8")
    else:
        print(report)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
```