#!/usr/bin/env python3
# Copyright (c) 2026 Christian Wirtz
# Licensed under the MIT License (see the LICENSE file).
"""Checkmk special agent for GitLab (self-hosted CE/EE).

Queries the GitLab REST API (v4) and emits Checkmk agent sections as JSON.
Design mirrors the built-in Jenkins special agent: one section per object type,
each carrying a JSON payload that the agent-based check plugins parse.

Sections produced:
    <<<gitlab_instance:sep(0)>>>   general instance / version information
    <<<gitlab_health:sep(0)>>>     /-/health, /-/readiness, /-/liveness results
    <<<gitlab_pipelines:sep(0)>>>  latest pipeline per selected project (+ref)
    <<<gitlab_jobs:sep(0)>>>       jobs of those latest pipelines
    <<<gitlab_runners:sep(0)>>>    runners visible to the token

The agent is intentionally dependency-free (standard library only) so it can be
shipped inside an MKP and executed on any Checkmk site without extra packages.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Iterable, Iterator
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any

USER_AGENT = "checkmk-gitlab-special-agent/1.0.0"
DEFAULT_TIMEOUT = 30
# Hard cap so a misconfigured "all projects" scope cannot run away.
MAX_PAGES = 50


@dataclass
class Args:
    url: str
    token: str | None
    health_token: str | None
    timeout: int
    no_cert_check: bool
    projects: list[str]
    project_specs: list[dict]
    membership: bool
    per_page: int
    ref: str | None
    project_display: str  # "full", "group_project" or "short"
    max_workers: int
    collect_jobs: bool
    collect_schedules: bool
    collect_schedule_duration: bool
    track_last_success: bool
    collect_runners: bool
    runner_scope: str  # "all" or "owned"
    collect_health: bool
    health_endpoints: list[str]
    check_updates: bool
    update_source: str  # "dockerhub" or "manual"
    update_target: str | None
    update_cache_ttl: int
    collect_license: bool
    collect_users: bool
    collect_ssh_keys: bool
    collect_statistics: bool
    debug: bool


def parse_arguments(argv: list[str]) -> Args:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", required=True,
                        help="Base URL of the GitLab server, e.g. https://gitlab.example.com")
    tok = parser.add_mutually_exclusive_group()
    tok.add_argument("--token", help="Personal/Project/Group access token (scope: read_api)")
    tok.add_argument("--token-file",
                     help="Read the access token from this file (alternative to --token)")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT,
                        help="HTTP timeout in seconds (default: %(default)s)")
    parser.add_argument("--health-token", default=None,
                        help="GitLab health-check token (passed as ?token=... to the "
                             "/-/health|readiness|liveness endpoints). This is the token "
                             "from GitLab Admin > Monitoring > Health Check, NOT the API "
                             "access token. If omitted, the endpoints are queried without "
                             "a token (works only if the caller IP is in GitLab's "
                             "monitoring allowlist).")
    parser.add_argument("--no-cert-check", action="store_true",
                        help="Do not verify the TLS certificate (self-signed setups)")
    parser.add_argument("--project", dest="projects", action="append", default=[],
                        help="Project ID or URL-encoded path. Repeatable. "
                             "If omitted, projects are discovered (see --membership).")
    parser.add_argument("--project-spec", dest="project_specs", action="append", default=[],
                        help="JSON object {\"project\": \"id-or-path\", \"ref\": \"branch\"} "
                             "selecting one pipeline to monitor (ref optional). Repeatable. "
                             "Takes precedence over discovery and limits the API calls to "
                             "exactly these projects (much faster than scanning all).")
    parser.add_argument("--membership", action="store_true",
                        help="When no project is given, only list projects the token "
                             "is a member of (instead of all accessible projects).")
    parser.add_argument("--per-page", type=int, default=50,
                        help="API page size (default: %(default)s, max 100)")
    parser.add_argument("--ref", default=None,
                        help="Restrict pipelines to this git ref/branch (e.g. main). "
                             "If omitted, the latest pipeline of any ref is used per project.")
    parser.add_argument("--max-workers", type=int, default=8,
                        help="Number of parallel API workers for per-project requests "
                             "(default: %(default)s, use 1 to disable parallelism).")
    parser.add_argument("--no-jobs", dest="collect_jobs", action="store_false", default=True,
                        help="Do not collect the jobs of the latest pipelines.")
    parser.add_argument("--no-schedules", dest="collect_schedules", action="store_false",
                        default=True, help="Do not collect pipeline schedules.")
    parser.add_argument("--no-schedule-duration", dest="collect_schedule_duration",
                        action="store_false", default=True,
                        help="Do not fetch the duration of each schedule's last pipeline "
                             "(saves one API call per schedule).")
    parser.add_argument("--track-last-success", action="store_true",
                        help="Report the last successful pipeline/job run and how long ago "
                             "it was (extra API calls when the latest run is not successful).")
    parser.add_argument("--project-display", choices=("full", "group_project", "short"),
                        default="short",
                        help="How the project appears in service names: full path, last two "
                             "path segments, or just the project name (default: short). The "
                             "full path and group are always available as service labels.")
    parser.add_argument("--no-runners", dest="collect_runners", action="store_false",
                        default=True, help="Do not collect runner information.")
    parser.add_argument("--runner-scope", choices=("all", "owned"), default="owned",
                        help="'all' needs an admin token (/runners/all); "
                             "'owned' uses /runners (default).")
    parser.add_argument("--no-health", dest="collect_health", action="store_false",
                        default=True, help="Do not query the monitoring endpoints.")
    parser.add_argument("--health-endpoint", dest="health_endpoints", action="append",
                        default=[], choices=("readiness", "liveness", "metrics"),
                        help="Monitoring endpoint under /-/ to query. Repeatable. "
                             "Defaults to readiness, liveness and metrics if not given.")
    parser.add_argument("--check-updates", action="store_true",
                        help="Compare the installed version against the latest available "
                             "release and report if an update is available.")
    parser.add_argument("--update-source", choices=("dockerhub", "manual"),
                        default="dockerhub",
                        help="Where to obtain the latest version: 'dockerhub' queries the "
                             "gitlab/gitlab-ee|ce image tags (needs internet on the Checkmk "
                             "server); 'manual' uses --update-target (for air-gapped setups).")
    parser.add_argument("--update-target", default=None,
                        help="Desired/latest version for --update-source manual, e.g. 17.11.1")
    parser.add_argument("--update-cache-ttl", type=int, default=28800,
                        help="Cache the looked-up latest version this many seconds "
                             "(default: %(default)s = 8h) to avoid frequent Docker Hub calls.")
    parser.add_argument("--collect-license", action="store_true",
                        help="Collect GitLab EE license info (needs an admin token; EE only).")
    parser.add_argument("--collect-users", action="store_true",
                        help="Collect user statistics (needs an admin token).")
    parser.add_argument("--collect-ssh-keys", action="store_true",
                        help="Collect SSH key statistics for the token's user "
                             "(active/expired counts and key types).")
    parser.add_argument("--collect-statistics", action="store_true",
                        help="Collect instance-wide application statistics "
                             "(GET /application/statistics; needs an admin token).")
    parser.add_argument("--debug", action="store_true",
                        help="Show tracebacks and raise on API errors.")

    args = parser.parse_args(argv)

    token = args.token
    if args.token_file:
        with open(args.token_file, encoding="utf-8") as handle:
            token = handle.read().strip()

    project_specs = []
    for raw_spec in args.project_specs:
        try:
            spec = json.loads(raw_spec)
        except (ValueError, TypeError):
            spec = {"project": raw_spec}
        if isinstance(spec, dict) and spec.get("project"):
            entry = {"project": str(spec["project"]), "ref": spec.get("ref") or None}
            if "collect_jobs" in spec:
                entry["collect_jobs"] = bool(spec["collect_jobs"])
            project_specs.append(entry)

    return Args(
        url=args.url.rstrip("/"),
        token=token,
        health_token=args.health_token,
        timeout=args.timeout,
        no_cert_check=args.no_cert_check,
        projects=args.projects,
        project_specs=project_specs,
        membership=args.membership,
        per_page=min(max(args.per_page, 1), 100),
        ref=args.ref,
        project_display=args.project_display,
        max_workers=max(1, args.max_workers),
        collect_jobs=args.collect_jobs,
        collect_schedules=args.collect_schedules,
        collect_schedule_duration=args.collect_schedule_duration,
        track_last_success=args.track_last_success,
        collect_runners=args.collect_runners,
        runner_scope=args.runner_scope,
        collect_health=args.collect_health,
        health_endpoints=args.health_endpoints or ["readiness", "liveness", "metrics"],
        check_updates=args.check_updates,
        update_source=args.update_source,
        update_target=args.update_target,
        update_cache_ttl=max(0, args.update_cache_ttl),
        collect_license=args.collect_license,
        collect_users=args.collect_users,
        collect_ssh_keys=args.collect_ssh_keys,
        collect_statistics=args.collect_statistics,
        debug=args.debug,
    )


class GitLabAPI:
    def __init__(self, args: Args) -> None:
        self._base = args.url
        self._api = f"{args.url}/api/v4"
        self._timeout = args.timeout
        self._debug = args.debug
        self._headers = {"User-Agent": USER_AGENT}
        if args.token:
            self._headers["PRIVATE-TOKEN"] = args.token
        self._health_token = args.health_token
        self._ctx = None
        if args.no_cert_check:
            import ssl
            self._ctx = ssl.create_default_context()
            self._ctx.check_hostname = False
            self._ctx.verify_mode = ssl.CERT_NONE

    def _open(self, url: str) -> tuple[int, dict[str, str], bytes]:
        request = urllib.request.Request(url, headers=self._headers)
        try:
            with urllib.request.urlopen(request, timeout=self._timeout, context=self._ctx) as resp:
                return resp.status, dict(resp.headers), resp.read()
        except urllib.error.HTTPError as exc:
            return exc.code, dict(exc.headers or {}), exc.read()

    def get_json(self, path: str, params: dict[str, Any] | None = None) -> Any:
        query = f"?{urllib.parse.urlencode(params)}" if params else ""
        url = f"{self._api}{path}{query}"
        status, _headers, body = self._open(url)
        if status >= 400:
            raise RuntimeError(f"GitLab API {path} returned HTTP {status}: {body[:200]!r}")
        return json.loads(body) if body else None

    def get_count(self, path: str, params: dict[str, Any] | None = None) -> int | None:
        """Return the total number of items via the X-Total header (per_page=1).

        Returns None when GitLab does not provide X-Total (it omits the header
        for result sets larger than 10,000 items), so callers must handle None.
        """
        query_params = dict(params or {})
        query_params["per_page"] = 1
        query_params["page"] = 1
        url = f"{self._api}{path}?{urllib.parse.urlencode(query_params)}"
        status, headers, body = self._open(url)
        if status >= 400:
            raise RuntimeError(f"GitLab API {path} returned HTTP {status}: {body[:200]!r}")
        total = None
        for key, value in headers.items():
            if key.lower() == "x-total":
                total = value
                break
        try:
            return int(total) if total is not None else None
        except (ValueError, TypeError):
            return None

    def get_paginated(self, path: str, params: dict[str, Any] | None = None,
                      per_page: int = 50) -> Iterator[Any]:
        page = 1
        params = dict(params or {})
        params["per_page"] = per_page
        while page <= MAX_PAGES:
            params["page"] = page
            query = f"?{urllib.parse.urlencode(params)}"
            url = f"{self._api}{path}{query}"
            status, headers, body = self._open(url)
            if status >= 400:
                raise RuntimeError(f"GitLab API {path} returned HTTP {status}: {body[:200]!r}")
            chunk = json.loads(body) if body else []
            if not chunk:
                break
            yield from chunk
            next_page = headers.get("X-Next-Page") or headers.get("x-next-page")
            if not next_page:
                break
            page = int(next_page)

    def get_health(self, endpoint: str) -> dict[str, Any]:
        url = f"{self._base}/-/{endpoint}"
        if self._health_token:
            url = f"{url}?{urllib.parse.urlencode({'token': self._health_token})}"
        # The monitoring endpoints authenticate via the ?token= query parameter (or an
        # IP allowlist), NOT via the API PRIVATE-TOKEN header. Sending the API token
        # here yields 404, so we deliberately omit it and send only the User-Agent.
        request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
        try:
            with urllib.request.urlopen(request, timeout=self._timeout, context=self._ctx) as resp:
                status = resp.status
                raw = resp.read()
        except urllib.error.HTTPError as exc:
            status = exc.code
            raw = exc.read()
        except (urllib.error.URLError, OSError) as exc:
            return {"http_status": None, "error": str(exc), "components": {}, "kind": "probe"}

        text = raw.decode("utf-8", "replace") if raw else ""

        # /-/metrics returns Prometheus exposition text, not JSON. Just verify it is
        # reachable and report how many metric samples were exposed.
        if endpoint == "metrics":
            sample_lines = sum(1 for line in text.splitlines()
                               if line and not line.startswith("#"))
            return {"http_status": status, "error": None, "kind": "metrics",
                    "sample_count": sample_lines, "components": {}}

        # readiness/liveness return JSON: {"status": "ok", "<comp>": [{"status": "ok"}], ...}
        components: dict[str, Any] = {}
        try:
            parsed = json.loads(text) if text else {}
            if isinstance(parsed, dict):
                for key, value in parsed.items():
                    if isinstance(value, list):
                        for entry in value:
                            comp_status = entry.get("status") if isinstance(entry, dict) else None
                            components[key] = comp_status or "unknown"
                if not components and isinstance(parsed.get("status"), str):
                    components["overall"] = parsed["status"]
        except (ValueError, TypeError):
            pass
        return {"http_status": status, "error": None, "components": components, "kind": "probe"}


def _project_ident(project: dict[str, Any]) -> str:
    return project.get("path_with_namespace") or str(project.get("id"))


def _project_fields(project: dict[str, Any], display: str) -> dict[str, Any]:
    """Return the item display name plus full path and group for labels."""
    full = project.get("path_with_namespace") or str(project.get("id"))
    parts = full.split("/")
    short = parts[-1]
    group = "/".join(parts[:-1]) if len(parts) > 1 else ""
    if display == "full":
        shown = full
    elif display == "group_project":
        shown = "/".join(parts[-2:]) if len(parts) > 1 else full
    else:  # "short"
        shown = short
    return {"project": shown, "project_full": full, "group": group}


def _encode_project(identifier: str) -> str:
    return urllib.parse.quote(identifier, safe="") if "/" in str(identifier) else str(identifier)


def _parallel_map(func, items: list, max_workers: int) -> list:
    """Run func over items, in parallel when max_workers > 1, preserving order."""
    items = list(items)
    if not items:
        return []
    if max_workers <= 1 or len(items) == 1:
        return [func(it) for it in items]
    with ThreadPoolExecutor(max_workers=min(max_workers, len(items))) as pool:
        return list(pool.map(func, items))


def resolve_targets(api: GitLabAPI, args: Args) -> list[tuple[dict[str, Any], str | None, bool | None]]:
    """Return the (project, ref, collect_jobs) triples to monitor.

    collect_jobs is the per-entry override (True/False) or None to fall back to
    the global --no-jobs setting. Explicit --project-spec / --project entries
    limit the work to exactly those projects (fast). Otherwise all
    accessible/membership projects are listed.
    """
    if args.project_specs:
        def _resolve(spec: dict) -> tuple[dict[str, Any], str | None, bool | None] | None:
            try:
                project = api.get_json(f"/projects/{_encode_project(spec['project'])}")
            except RuntimeError:
                return None
            return (project, spec.get("ref") or args.ref, spec.get("collect_jobs"))
        return [t for t in _parallel_map(_resolve, args.project_specs, args.max_workers)
                if t is not None]

    if args.projects:
        def _resolve_plain(identifier: str) -> tuple[dict[str, Any], str | None, bool | None] | None:
            try:
                return (api.get_json(f"/projects/{_encode_project(identifier)}"), args.ref, None)
            except RuntimeError:
                return None
        return [t for t in _parallel_map(_resolve_plain, args.projects, args.max_workers)
                if t is not None]

    params: dict[str, Any] = {"archived": "false", "order_by": "last_activity_at"}
    if args.membership:
        params["membership"] = "true"
    projects = list(api.get_paginated("/projects", params, per_page=args.per_page))
    return [(project, args.ref, None) for project in projects]


def collect_pipelines_and_jobs(
    api: GitLabAPI, args: Args, targets: list[tuple[dict[str, Any], str | None, bool | None]]
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    def _one(target: tuple[dict[str, Any], str | None, bool | None]):
        project, ref, want_jobs = target
        pid = project["id"]
        pf = _project_fields(project, args.project_display)
        ident = pf["project"]
        # Per-entry override wins; otherwise fall back to the global setting.
        collect_jobs = args.collect_jobs if want_jobs is None else want_jobs
        params = {"ref": ref} if ref else None
        try:
            latest = api.get_json(f"/projects/{pid}/pipelines/latest", params)
        except RuntimeError:
            latest = None
        if not latest:
            return None, []
        pipeline = {
            "project": ident,
            "project_full": pf["project_full"],
            "group": pf["group"],
            "project_id": pid,
            "pipeline_id": latest.get("id"),
            "ref": latest.get("ref"),
            "status": latest.get("status"),
            "detailed_status_group": (latest.get("detailed_status") or {}).get("group"),
            "detailed_status_label": (latest.get("detailed_status") or {}).get("label"),
            "source": latest.get("source"),
            "sha": (latest.get("sha") or "")[:12],
            "web_url": latest.get("web_url"),
            "created_at": latest.get("created_at"),
            "updated_at": latest.get("updated_at"),
            "duration": latest.get("duration"),
        }

        # Optional: when did the pipeline last succeed? Reuse the latest run if it
        # already succeeded, otherwise look back once (cheap: newest success only).
        if args.track_last_success:
            if latest.get("status") == "success":
                pipeline["last_success_at"] = latest.get("updated_at") or latest.get("created_at")
            else:
                sp = {"status": "success", "per_page": 1}
                if ref:
                    sp["ref"] = ref
                try:
                    succ = list(api.get_paginated(f"/projects/{pid}/pipelines", sp, per_page=1))
                except RuntimeError:
                    succ = []
                pipeline["last_success_at"] = (
                    succ[0].get("updated_at") or succ[0].get("created_at")) if succ else None

        job_records: list[dict[str, Any]] = []
        if collect_jobs:
            try:
                pipeline_jobs = list(api.get_paginated(
                    f"/projects/{pid}/pipelines/{latest['id']}/jobs",
                    {"per_page": args.per_page}, per_page=args.per_page))
            except RuntimeError:
                pipeline_jobs = []

            # Map job name -> most recent successful finished_at (one call per project),
            # only needed when a monitored job is not currently successful.
            last_success_by_name: dict[str, Any] = {}
            if args.track_last_success and any(
                    j.get("status") != "success" for j in pipeline_jobs):
                try:
                    succ_jobs = list(api.get_paginated(
                        f"/projects/{pid}/jobs", {"scope": "success", "per_page": 100},
                        per_page=100))
                except RuntimeError:
                    succ_jobs = []
                for sj in succ_jobs:
                    name = sj.get("name")
                    if name and name not in last_success_by_name:
                        last_success_by_name[name] = sj.get("finished_at")

            for job in pipeline_jobs:
                record = {
                    "project": ident,
                    "project_full": pf["project_full"],
                    "group": pf["group"],
                    "project_id": pid,
                    "pipeline_id": latest.get("id"),
                    "job_id": job.get("id"),
                    "name": job.get("name"),
                    "stage": job.get("stage"),
                    "status": job.get("status"),
                    "ref": job.get("ref"),
                    "allow_failure": bool(job.get("allow_failure")),
                    "duration": job.get("duration"),
                    "web_url": job.get("web_url"),
                }
                if args.track_last_success:
                    if job.get("status") == "success":
                        record["last_success_at"] = job.get("finished_at")
                    else:
                        record["last_success_at"] = last_success_by_name.get(job.get("name"))
                job_records.append(record)
        return pipeline, job_records

    pipelines: list[dict[str, Any]] = []
    jobs: list[dict[str, Any]] = []
    for pipeline, job_records in _parallel_map(_one, targets, args.max_workers):
        if pipeline is not None:
            pipelines.append(pipeline)
            jobs.extend(job_records)
    return pipelines, jobs


def collect_runners(api: GitLabAPI, args: Args) -> list[dict[str, Any]]:
    path = "/runners/all" if args.runner_scope == "all" else "/runners"
    listed = list(api.get_paginated(path, per_page=args.per_page))

    def _detail(runner: dict[str, Any]) -> dict[str, Any]:
        # /runners returns a summary; fetch details for contacted_at/ip when possible.
        detail = runner
        try:
            detail = api.get_json(f"/runners/{runner['id']}")
        except RuntimeError:
            pass
        return {
            "id": detail.get("id"),
            "description": detail.get("description") or "",
            "runner_type": detail.get("runner_type"),
            "active": detail.get("active", detail.get("paused") is False),
            "paused": detail.get("paused"),
            "online": detail.get("online"),
            "status": detail.get("status"),
            "ip_address": detail.get("ip_address"),
            "contacted_at": detail.get("contacted_at"),
            "version": detail.get("version"),
        }

    return _parallel_map(_detail, listed, args.max_workers)


def collect_pipeline_schedules(
    api: GitLabAPI, args: Args, projects: Iterable[dict[str, Any]]
) -> list[dict[str, Any]]:
    def _one(project: dict[str, Any]) -> list[dict[str, Any]]:
        pid = project["id"]
        pf = _project_fields(project, args.project_display)
        ident = pf["project"]
        try:
            entries = list(api.get_paginated(
                f"/projects/{pid}/pipeline_schedules", per_page=args.per_page))
        except RuntimeError:
            return []
        out: list[dict[str, Any]] = []
        for entry in entries:
            # The list endpoint omits the last pipeline; the detail endpoint has it.
            detail = entry
            try:
                detail = api.get_json(
                    f"/projects/{pid}/pipeline_schedules/{entry['id']}")
            except RuntimeError:
                pass
            last_pipeline = detail.get("last_pipeline") or {}
            last_id = last_pipeline.get("id")
            last_status = last_pipeline.get("status")
            last_duration = last_pipeline.get("duration")
            # The schedule's last_pipeline usually omits the duration -> fetch the
            # pipeline once to obtain it (and an authoritative status), unless the
            # extra call was disabled.
            if last_id and last_duration is None and args.collect_schedule_duration:
                try:
                    pl = api.get_json(f"/projects/{pid}/pipelines/{last_id}")
                    last_duration = pl.get("duration")
                    last_status = pl.get("status") or last_status
                except RuntimeError:
                    pass
            failed_jobs: list[str] = []
            warning_jobs: list[str] = []
            # Only when the last scheduled run had a problem do we spend one extra
            # call to find out which job(s) were responsible.
            if last_id and last_status in ("failed", "canceled"):
                try:
                    pljobs = list(api.get_paginated(
                        f"/projects/{pid}/pipelines/{last_id}/jobs",
                        {"per_page": args.per_page}, per_page=args.per_page))
                except RuntimeError:
                    pljobs = []
                for job in pljobs:
                    if job.get("status") == "failed":
                        if job.get("allow_failure"):
                            warning_jobs.append(job.get("name"))
                        else:
                            failed_jobs.append(job.get("name"))
            out.append({
                "project": ident,
                "project_full": pf["project_full"],
                "group": pf["group"],
                "project_id": pid,
                "schedule_id": detail.get("id"),
                "description": detail.get("description") or "",
                "ref": detail.get("ref"),
                "cron": detail.get("cron"),
                "cron_timezone": detail.get("cron_timezone"),
                "next_run_at": detail.get("next_run_at"),
                "active": bool(detail.get("active")),
                "owner": (detail.get("owner") or {}).get("username"),
                "last_pipeline_id": last_id,
                "last_pipeline_status": last_status,
                "last_duration": last_duration,
                "failed_jobs": failed_jobs,
                "warning_jobs": warning_jobs,
            })
        return out

    schedules: list[dict[str, Any]] = []
    for chunk in _parallel_map(_one, list(projects), args.max_workers):
        schedules.extend(chunk)
    return schedules


def collect_license(api: GitLabAPI) -> dict[str, Any] | None:
    """GET /license (GitLab EE, admin token). Returns None if unavailable."""
    try:
        lic = api.get_json("/license")
    except RuntimeError:
        return None
    if not isinstance(lic, dict) or not lic:
        return None
    return {
        "plan": lic.get("plan"),
        "licensee": (lic.get("licensee") or {}).get("Name") or (lic.get("licensee") or {}).get("Company"),
        "starts_at": lic.get("starts_at"),
        "expires_at": lic.get("expires_at"),
        "expired": lic.get("expired"),
        "user_limit": lic.get("user_limit"),
        "active_users": lic.get("active_users"),
        "historical_max": lic.get("historical_max"),
        "overage": lic.get("overage"),
    }


def collect_users(api: GitLabAPI) -> dict[str, Any] | None:
    """User statistics via counting queries (X-Total header), not by downloading
    the whole user list — this is fast even on large instances.

    GitLab omits X-Total for result sets > 10,000, in which case the affected
    counts are reported as null and the check shows them as unavailable.
    """
    try:
        total = api.get_count("/users")
    except RuntimeError:
        return None

    if total is None:
        # X-Total not provided (e.g. gitlab.com's huge public user base).
        return {"count_unavailable": True}

    def _count(params: dict[str, Any]) -> int | None:
        try:
            return api.get_count("/users", params)
        except RuntimeError:
            return None

    humans = _count({"exclude_internal": "true"})
    active = _count({"active": "true"})
    billable = _count({"active": "true", "exclude_internal": "true"})
    blocked = _count({"blocked": "true"})

    bots = (total - humans) if (humans is not None) else None
    deactivated = None
    if active is not None and blocked is not None:
        deactivated = max(0, total - active - blocked)

    return {
        "count_unavailable": False,
        "total": total,
        "humans": humans,
        "bots": bots,
        "active": active,          # active (any account, incl. active bots)
        "blocked": blocked,
        "deactivated": deactivated,  # approx: total - active - blocked
        "billable": billable,      # active AND non-internal (bot) == GitLab billable-ish
    }


def _ssh_key_type(key_blob: str | None) -> str:
    """Map the leading token of an OpenSSH public key to a normalized type."""
    if not key_blob:
        return "unknown"
    prefix = key_blob.strip().split(None, 1)[0].lower()
    mapping = {
        "ssh-rsa": "rsa",
        "ssh-ed25519": "ed25519",
        "ssh-dss": "dsa",
        "sk-ssh-ed25519@openssh.com": "ed25519_sk",
    }
    if prefix in mapping:
        return mapping[prefix]
    if prefix.startswith("ecdsa-sha2-"):
        return "ecdsa"
    if prefix.startswith("sk-ecdsa-sha2-"):
        return "ecdsa_sk"
    return "other"


# Key types are classified in the check plugin; which ones count as deprecated
# is configurable there (e.g. DSA, optionally RSA).


def _rsa_key_bits(key_blob: str | None) -> int | None:
    """Return the RSA modulus size in bits for an 'ssh-rsa' key, else None.

    Parses the base64 SSH wire format (string algo, mpint e, mpint n) using only
    the standard library. Only the size is derived; the key material is not kept.
    """
    if not key_blob:
        return None
    parts = key_blob.strip().split()
    if len(parts) < 2 or parts[0] != "ssh-rsa":
        return None
    import base64
    try:
        data = base64.b64decode(parts[1])
    except Exception:  # noqa: BLE001
        return None

    def _read(offset: int) -> tuple[bytes, int]:
        length = int.from_bytes(data[offset:offset + 4], "big")
        offset += 4
        return data[offset:offset + length], offset + length

    try:
        offset = 0
        _algo, offset = _read(offset)
        _e, offset = _read(offset)
        modulus, offset = _read(offset)
    except (IndexError, ValueError):
        return None
    modulus = modulus.lstrip(b"\x00")
    if not modulus:
        return None
    return len(modulus) * 8


def collect_ssh_keys(api: GitLabAPI, args: Args) -> dict[str, Any] | None:
    """Instance-wide SSH key statistics (admin token).

    GitLab has no bulk 'all SSH keys' API, so this lists users and aggregates
    each user's keys (GET /users/:id/keys). This is an administrator, self-hosted
    feature and can be slow on large instances; per-user requests run in
    parallel. Only key metadata is collected, not the key material.
    """
    from datetime import datetime as _dt, timezone as _tz

    try:
        users = list(api.get_paginated("/users", {"per_page": args.per_page},
                                       per_page=args.per_page))
    except RuntimeError:
        return None
    if not users:
        return None

    def _user_keys(user: dict[str, Any]) -> list[dict[str, Any]]:
        uid = user.get("id")
        if uid is None:
            return []
        try:
            return list(api.get_paginated(f"/users/{uid}/keys", {"per_page": 100},
                                          per_page=100))
        except RuntimeError:
            return []

    now = _dt.now(_tz.utc)
    total = active = expired = 0
    users_with_keys = 0
    by_type: dict[str, int] = {}
    rsa_bits: list[int] = []
    for keys in _parallel_map(_user_keys, users, args.max_workers):
        if keys:
            users_with_keys += 1
        for key in keys:
            total += 1
            blob = key.get("key")
            ktype = _ssh_key_type(blob)
            by_type[ktype] = by_type.get(ktype, 0) + 1
            if ktype == "rsa":
                bits = _rsa_key_bits(blob)
                if bits is not None:
                    rsa_bits.append(bits)
            expires = key.get("expires_at")
            is_expired = False
            if expires:
                try:
                    exp = _dt.fromisoformat(str(expires).replace("Z", "+00:00"))
                    is_expired = exp < now
                except (ValueError, TypeError):
                    is_expired = False
            if is_expired:
                expired += 1
            else:
                active += 1

    return {
        "users_scanned": len(users),
        "users_with_keys": users_with_keys,
        "total": total,
        "active": active,
        "expired": expired,
        "by_type": by_type,
        "rsa_bits": sorted(rsa_bits),
        "truncated": len(users) >= MAX_PAGES * args.per_page,
    }


def collect_statistics(api: GitLabAPI) -> dict[str, Any] | None:
    """Instance-wide counts from GET /application/statistics (admin token).

    The API returns the numbers as strings in some versions, so values are
    coerced to int. See https://docs.gitlab.com/api/statistics/
    """
    try:
        stats = api.get_json("/application/statistics")
    except RuntimeError:
        return None
    if not isinstance(stats, dict) or not stats:
        return None

    fields = ("forks", "issues", "merge_requests", "notes", "snippets", "ssh_keys",
              "milestones", "users", "projects", "groups", "active_users")
    out: dict[str, Any] = {}
    for field in fields:
        value = stats.get(field)
        if value is None:
            continue
        try:
            out[field] = int(str(value).replace(",", "").strip())
        except (ValueError, TypeError):
            continue
    return out or None


def _emit(section: str, payload: Any) -> None:
    sys.stdout.write(f"<<<{section}:sep(0)>>>\n")
    sys.stdout.write(json.dumps(payload, default=str))
    sys.stdout.write("\n")


_SEMVER_RE = re.compile(r"(\d+)\.(\d+)\.(\d+)")


def _parse_semver(value: str | None) -> tuple[int, int, int] | None:
    if not value:
        return None
    match = _SEMVER_RE.search(str(value))
    if not match:
        return None
    return (int(match.group(1)), int(match.group(2)), int(match.group(3)))


def _edition_of(version: str | None) -> str:
    return "ce" if version and "-ce" in version else "ee"


def _http_get_json(url: str, timeout: int) -> Any:
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT,
                                                   "Accept": "application/json"})
    with urllib.request.urlopen(request, timeout=timeout) as resp:
        return json.loads(resp.read())


def _dockerhub_latest(edition: str, timeout: int, max_pages: int = 5) -> str | None:
    """Return the highest stable gitlab/gitlab-<edition> image version on Docker Hub."""
    pattern = re.compile(rf"^(\d+)\.(\d+)\.(\d+)-{edition}(?:\.\d+)?$")
    best: tuple[int, int, int] | None = None
    best_str: str | None = None
    url = (f"https://hub.docker.com/v2/repositories/gitlab/gitlab-{edition}"
           f"/tags?page_size=100&ordering=last_updated")
    for _page in range(max_pages):
        data = _http_get_json(url, timeout)
        for entry in data.get("results", []):
            name = entry.get("name", "")
            m = pattern.match(name)
            if not m:
                continue
            ver = (int(m.group(1)), int(m.group(2)), int(m.group(3)))
            if best is None or ver > best:
                best, best_str = ver, f"{ver[0]}.{ver[1]}.{ver[2]}"
        url = data.get("next")
        if not url:
            break
    return best_str


def _cache_path(edition: str) -> str:
    base = os.environ.get("OMD_ROOT")
    base = os.path.join(base, "tmp") if base else tempfile.gettempdir()
    return os.path.join(base, f"agent_gitlab_latest_{edition}.json")


def _read_cache(edition: str, ttl: int) -> dict[str, Any] | None:
    path = _cache_path(edition)
    try:
        if ttl > 0 and (time.time() - os.path.getmtime(path)) <= ttl:
            with open(path, encoding="utf-8") as handle:
                return json.load(handle)
    except (OSError, ValueError):
        return None
    return None


def _write_cache(edition: str, payload: dict[str, Any]) -> None:
    try:
        with open(_cache_path(edition), "w", encoding="utf-8") as handle:
            json.dump(payload, handle)
    except OSError:
        pass


def build_update_info(args: Args, version_str: str | None) -> dict[str, Any]:
    installed = _parse_semver(version_str)
    info: dict[str, Any] = {
        "enabled": True,
        "source": args.update_source,
        "installed": f"{installed[0]}.{installed[1]}.{installed[2]}" if installed else None,
        "latest": None,
        "update_available": None,
        "error": None,
    }

    if args.update_source == "manual":
        info["latest"] = (args.update_target or "").strip() or None
        if info["latest"] is None:
            info["error"] = "manual update source selected but no target version given"
    else:  # dockerhub
        edition = _edition_of(version_str)
        info["edition"] = edition
        cached = _read_cache(edition, args.update_cache_ttl)
        if cached and cached.get("latest"):
            info["latest"] = cached["latest"]
            info["cached"] = True
        else:
            try:
                latest = _dockerhub_latest(edition, args.timeout)
                info["latest"] = latest
                if latest:
                    _write_cache(edition, {"latest": latest})
                else:
                    info["error"] = "no matching stable image tag found on Docker Hub"
            except Exception as exc:  # noqa: BLE001 - never break the agent on this
                info["error"] = f"Docker Hub lookup failed: {exc}"

    latest = _parse_semver(info["latest"])
    if installed is not None and latest is not None:
        info["update_available"] = latest > installed
    return info


def agent_gitlab_main(args: Args) -> int:
    api = GitLabAPI(args)

    # Instance / version (requires a token; degrade gracefully otherwise).
    try:
        version = api.get_json("/version")
    except RuntimeError as exc:
        version = {"error": str(exc)}
    instance: dict[str, Any] = {"url": args.url, "version": version}
    if args.check_updates:
        version_str = version.get("version") if isinstance(version, dict) else None
        instance["update"] = build_update_info(args, version_str)
    _emit("gitlab_instance", instance)

    if args.collect_health:
        health = {name: api.get_health(name) for name in args.health_endpoints}
        _emit("gitlab_health", health)

    targets = resolve_targets(api, args)
    pipelines, jobs = collect_pipelines_and_jobs(api, args, targets)
    _emit("gitlab_pipelines", pipelines)
    # Emit the jobs section if jobs were collected globally or for any single entry.
    if args.collect_jobs or jobs:
        _emit("gitlab_jobs", jobs)

    if args.collect_schedules:
        try:
            projects = [project for project, _ref, _cj in targets]
            schedules = collect_pipeline_schedules(api, args, projects)
        except RuntimeError as exc:
            if args.debug:
                raise
            schedules = []
            sys.stderr.write(f"Schedule collection failed: {exc}\n")
        _emit("gitlab_pipeline_schedules", schedules)

    if args.collect_runners:
        try:
            runners = collect_runners(api, args)
        except RuntimeError as exc:
            if args.debug:
                raise
            runners = []
            sys.stderr.write(f"Runner collection failed: {exc}\n")
        _emit("gitlab_runners", runners)

    if args.collect_license:
        _emit("gitlab_license", collect_license(api) or {})

    if args.collect_users:
        _emit("gitlab_users", collect_users(api) or {})

    if args.collect_ssh_keys:
        _emit("gitlab_ssh_keys", collect_ssh_keys(api, args) or {})

    if args.collect_statistics:
        _emit("gitlab_statistics", collect_statistics(api) or {})

    return 0


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(sys.argv[1:] if argv is None else argv)
    try:
        return agent_gitlab_main(args)
    except Exception as exc:  # noqa: BLE001 - agents must fail with a clean message
        if args.debug:
            raise
        sys.stderr.write(f"agent_gitlab: {exc}\n")
        return 1


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