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

Queries the GitLab REST API (v4) and emits Checkmk agent sections as JSON.
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
    <<<gitlab_mirrors:sep(0)>>>    push mirrors of the projects (optional, incl. target check)

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
    collect_status_page: bool
    status_page_id: str
    collect_mirrors: bool = False
    mirror_cache_ttl: int = 0
    verify_mirror_target: bool = False
    mirror_target_cache_ttl: int = 3600
    mirror_target_tags: bool = False
    mirror_target_token: str | None = None
    mirror_target_hosts: tuple[str, ...] = ("gitlab.com",)
    debug: bool = False


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("--collect-status-page", action="store_true",
                        help="Collect the public GitLab.com status page "
                             "(https://status.gitlab.com, via the status.io API; no token).")
    parser.add_argument("--status-page-id", default=GITLAB_STATUS_PAGE_ID,
                        help="status.io page id of the status page (default: %(default)s "
                             "= status.gitlab.com).")
    parser.add_argument("--collect-mirrors", action="store_true",
                        help="Collect the push mirrors of the monitored projects "
                             "(GET /projects/:id/remote_mirrors, one call per project; "
                             "needs at least the Maintainer role).")
    parser.add_argument("--mirror-cache-ttl", type=int, default=0,
                        help="Cache the mirror status this many seconds (default: 0 = "
                             "query on every run).")
    parser.add_argument("--verify-mirror-target", action="store_true",
                        help="Also query each mirror's target GitLab (archived? default "
                             "branch on the same commit?). Needs --collect-mirrors.")
    parser.add_argument("--mirror-target-cache-ttl", type=int, default=3600,
                        help="Cache an in-sync target result this many seconds "
                             "(default: %(default)s). A changed source invalidates it.")
    parser.add_argument("--mirror-target-tags", action="store_true",
                        help="Also compare the tags between source and target.")
    parser.add_argument("--mirror-target-token", default=None,
                        help="Access token for the target GitLab (only needed for "
                             "private targets).")
    parser.add_argument("--mirror-target-host", action="append", default=None,
                        dest="mirror_target_hosts",
                        help="Mirror target host that may be queried (repeatable; "
                             "default: gitlab.com). Targets on other hosts are not "
                             "contacted, so the target token never leaves these hosts.")
    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,
        collect_status_page=args.collect_status_page,
        status_page_id=args.status_page_id,
        collect_mirrors=args.collect_mirrors or args.verify_mirror_target,
        mirror_cache_ttl=max(0, args.mirror_cache_ttl),
        verify_mirror_target=args.verify_mirror_target,
        mirror_target_cache_ttl=max(0, args.mirror_target_cache_ttl),
        mirror_target_tags=args.mirror_target_tags,
        mirror_target_token=args.mirror_target_token,
        mirror_target_hosts=tuple(h.strip().lower() for h in
                                  (args.mirror_target_hosts or ["gitlab.com"])
                                  if h.strip()),
        debug=args.debug,
    )


# status.gitlab.com is hosted on status.io; this is its page id (see the page's
# "x-status-page-id" response header).
GITLAB_STATUS_PAGE_ID = "5b36dc6502d06804c08349f7"
STATUS_IO_API = "https://api.status.io/1.0/status/{page_id}"


class RateLimited(RuntimeError):
    """HTTP 429 from the API; retry_after is in seconds."""

    def __init__(self, message: str, retry_after: int) -> None:
        super().__init__(message)
        self.retry_after = retry_after


def _raise_for_status(path: str, status: int, headers: dict[str, str], body: bytes) -> None:
    if status == 429:
        lowered = {k.lower(): v for k, v in headers.items()}
        try:
            retry = int(lowered.get("retry-after", "60"))
        except ValueError:
            retry = 60
        raise RateLimited(f"GitLab API {path} rate-limited (HTTP 429, retry after {retry}s)",
                          retry)
    if status >= 400:
        raise RuntimeError(f"GitLab API {path} returned HTTP {status}: {body[:200]!r}")


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)
        _raise_for_status(path, status, headers, body)
        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)
            _raise_for_status(path, status, headers, body)
            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))
    # Projects scheduled for deletion (renamed to "<name>-deletion_scheduled-<id>"
    # by GitLab) are about to disappear - don't create services for them.
    return [(project, args.ref, None) for project in projects
            if not _marked_for_deletion(project)]


def _marked_for_deletion(project: dict[str, Any]) -> bool:
    return bool(project.get("marked_for_deletion_on")
                or project.get("marked_for_deletion_at"))


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}
                # Without an explicit ref, /pipelines/latest is the default branch:
                # restrict the lookup to that branch, or a tag pipeline would count.
                success_ref = ref or latest.get("ref") or project.get("default_branch")
                if success_ref:
                    sp["ref"] = success_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


_MIRROR_URL_RE = re.compile(
    r"^(?P<scheme>[a-z+]+)://(?:[^@/]*@)?(?P<host>[^/:]+)(?::\d+)?/(?P<path>.+?)(?:\.git)?/?$")


def parse_mirror_url(url: str | None) -> dict[str, str] | None:
    """Split a mirror URL (ssh://git@host/a/b.git, https://user:***@host/a/b.git)
    into host and project path; credentials are dropped."""
    match = _MIRROR_URL_RE.match((url or "").strip())
    if not match:
        return None
    return {"host": match.group("host").lower(), "path": match.group("path")}


def _state_file(prefix: str, key: str) -> str:
    import hashlib
    base = os.environ.get("OMD_ROOT")
    base = os.path.join(base, "tmp") if base else tempfile.gettempdir()
    digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:16]
    return os.path.join(base, f"agent_gitlab_{prefix}_{digest}.json")


def _cache_get(path: str, ttl: int) -> Any:
    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):
        pass
    return None


def _cache_put(path: str, payload: Any) -> None:
    try:
        with open(path, "w", encoding="utf-8") as handle:
            json.dump(payload, handle)
    except OSError:
        pass


def _ref_map(api: "GitLabAPI", pid: Any, kind: str) -> dict[str, str]:
    """name -> commit sha for all branches or tags of a project."""
    return {entry.get("name"): (entry.get("commit") or {}).get("id")
            for entry in api.get_paginated(f"/projects/{pid}/repository/{kind}",
                                           per_page=100)}


def _source_refs(api: "GitLabAPI", project: dict[str, Any], want_tags: bool) -> dict[str, Any]:
    pid = project["id"]
    branch = project.get("default_branch")
    state: dict[str, Any] = {"branch": branch, "commit": None, "tags": None}
    if branch:
        state["commit"] = (api.get_json(
            f"/projects/{pid}/repository/branches/{urllib.parse.quote(branch, safe='')}")
            .get("commit") or {}).get("id")
    if want_tags:
        state["tags"] = _ref_map(api, pid, "tags")
    return state


def _query_target(api: "GitLabAPI", enc: str, source: dict[str, Any]) -> dict[str, Any]:
    """Read the target project and compare its refs with the source state."""
    result: dict[str, Any] = {}
    project = api.get_json(f"/projects/{enc}")
    result["archived"] = bool(project.get("archived"))
    result["builds_access_level"] = project.get("builds_access_level")
    branch = source.get("branch")
    result["branch"] = branch
    if branch:
        try:
            tgt = api.get_json(f"/projects/{enc}/repository/branches/"
                               f"{urllib.parse.quote(branch, safe='')}")
            result["target_commit"] = (tgt.get("commit") or {}).get("id")
        except RateLimited:
            raise
        except RuntimeError:  # branch missing on the target
            result["target_commit"] = None
        result["source_commit"] = source.get("commit")
        result["branch_in_sync"] = result["target_commit"] == source.get("commit")
    if source.get("tags") is not None:
        tags = _ref_map(api, enc, "tags")
        src_tags = source["tags"]
        result["tags_source"] = len(src_tags)
        result["tags_target"] = len(tags)
        result["tags_missing"] = sorted(t for t in src_tags if t not in tags)
        result["tags_differ"] = sorted(t for t in src_tags
                                       if t in tags and tags[t] != src_tags[t])
        result["tags_extra"] = sorted(t for t in tags if t not in src_tags)
    return result


# At most this many concurrent requests against a mirror target (e.g. gitlab.com,
# which rate-limits unauthenticated API clients aggressively).
import threading  # noqa: E402
_TARGET_SLOTS = threading.Semaphore(2)


def _target_blocked_until(host: str) -> float:
    data = _cache_get(_state_file("mirror_block", host), 86400)
    return float(data.get("until", 0)) if isinstance(data, dict) else 0.0


def _block_target(host: str, seconds: int) -> None:
    _cache_put(_state_file("mirror_block", host),
               {"until": time.time() + max(60, min(seconds, 86400))})


def _stale_or(result: dict[str, Any], error: str, last_file: str,
              fingerprint: str) -> dict[str, Any]:
    """Return the last real result for the same source state (marked stale), else
    the error result."""
    last = _cache_get(last_file, 7 * 86400)
    if isinstance(last, dict) and last.get("fingerprint") == fingerprint:
        stale = dict(last["result"])
        stale.update({"stale": True, "stale_reason": error[:300], "cached": True,
                      "rate_limited": bool(result.get("rate_limited"))})
        return stale
    result["error"] = error[:300]
    return result


def verify_mirror_target(args: Args, target: dict[str, str],
                         source: dict[str, Any]) -> dict[str, Any]:
    """Query the mirror target and compare it with the source state.

    An in-sync result is cached (key includes the source state, so a new push on
    the source invalidates it); mismatches and errors are re-checked next run.
    """
    # The mirror URL is set by project Maintainers. Only contact allowed hosts,
    # otherwise the target token would be sent to (and the site would make requests
    # to) any host a Maintainer enters.
    if target["host"] not in args.mirror_target_hosts:
        return {"checked": True, "cached": False,
                "error": f"target host {target['host']} is not in the allowed mirror "
                         f"target hosts ({', '.join(args.mirror_target_hosts) or 'none'})"}
    import hashlib
    fingerprint = hashlib.sha256(json.dumps(source, sort_keys=True).encode()).hexdigest()
    cache_file = _state_file("mirror_target", f"{target['host']}/{target['path']}")
    last_file = cache_file.replace(".json", "_last.json")
    cached = _cache_get(cache_file, args.mirror_target_cache_ttl)
    if isinstance(cached, dict) and cached.get("fingerprint") == fingerprint:
        result = dict(cached["result"])
        result["cached"] = True
        return result

    target_args = Args(**{**args.__dict__, "url": f"https://{target['host']}",
                          "token": args.mirror_target_token})
    api = GitLabAPI(target_args)
    enc = urllib.parse.quote(target["path"], safe="")
    result: dict[str, Any] = {"checked": True, "error": None, "cached": False,
                              "checked_at": int(time.time())}
    try:
        blocked = _target_blocked_until(target["host"])
        if blocked > time.time():
            raise RateLimited(f"{target['host']} rate limit: no requests until "
                              f"{time.strftime('%H:%M:%S', time.localtime(blocked))}",
                              int(blocked - time.time()))
        with _TARGET_SLOTS:
            result.update(_query_target(api, enc, source))
    except RateLimited as exc:
        if _target_blocked_until(target["host"]) <= time.time():
            _block_target(target["host"], exc.retry_after)
        result["rate_limited"] = True
        return _stale_or(result, str(exc), last_file, fingerprint)
    except (RuntimeError, urllib.error.URLError, OSError, ValueError) as exc:
        return _stale_or(result, str(exc), last_file, fingerprint)

    in_sync = (result.get("branch_in_sync", True) and not result.get("archived")
               and not result.get("tags_missing") and not result.get("tags_differ")
               and not result.get("tags_extra"))
    _cache_put(last_file, {"fingerprint": fingerprint, "result": result})
    if in_sync:
        _cache_put(cache_file, {"fingerprint": fingerprint, "result": result})
    return result


def collect_mirrors(api: GitLabAPI, args: Args,
                    projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Push mirrors of the given projects (one remote_mirrors call per project)."""
    cache_file = _state_file("mirrors", args.url + "|" + ",".join(
        sorted(str(p.get("id")) for p in projects)))
    raw: list[tuple[dict[str, Any], list[dict[str, Any]]]] | None = None
    cached = _cache_get(cache_file, args.mirror_cache_ttl)
    if isinstance(cached, list):
        by_id = {p.get("id"): p for p in projects}
        raw = [(by_id[pid], mirrors) for pid, mirrors in cached if pid in by_id]

    if raw is None:
        def _one(project: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
            try:
                return project, list(api.get_json(f"/projects/{project['id']}/remote_mirrors")
                                     or [])
            except RuntimeError:  # 403 without Maintainer role, feature disabled, ...
                return project, []
        raw = _parallel_map(_one, projects, args.max_workers)
        if args.mirror_cache_ttl > 0:
            _cache_put(cache_file, [(p["id"], m) for p, m in raw])

    def _record(entry: tuple[dict[str, Any], dict[str, Any]]) -> dict[str, Any]:
        project, mirror = entry
        pf = _project_fields(project, args.project_display)
        target = parse_mirror_url(mirror.get("url"))
        record = {
            **pf,
            "project_id": project.get("id"),
            "mirror_id": mirror.get("id"),
            "target": f"{target['host']}/{target['path']}" if target else
                      re.sub(r"//[^@/]*@", "//", str(mirror.get("url") or "")),
            "enabled": bool(mirror.get("enabled")),
            "update_status": mirror.get("update_status"),
            "last_error": mirror.get("last_error"),
            "last_update_at": mirror.get("last_update_at"),
            "last_update_started_at": mirror.get("last_update_started_at"),
            "last_successful_update_at": mirror.get("last_successful_update_at"),
            "only_protected_branches": bool(mirror.get("only_protected_branches")),
            "auth_method": mirror.get("auth_method"),
        }
        if args.verify_mirror_target and record["enabled"]:
            if target is None:
                record["target_check"] = {"checked": True, "error": "cannot parse mirror URL"}
            else:
                try:
                    source = _source_refs(api, project, args.mirror_target_tags)
                except RuntimeError as exc:
                    record["target_check"] = {"checked": True,
                                              "error": f"source lookup failed: {exc}"[:300]}
                else:
                    record["target_check"] = verify_mirror_target(args, target, source)
        return record

    entries = [(p, m) for p, mirrors in raw for m in mirrors if isinstance(m, dict)]
    return _parallel_map(_record, entries, args.max_workers)


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=true" (GitLab 16+) excludes every non-human account: project/group
    # bots, service accounts, internal users (GitLab Duo, ghost, ...) and the
    # placeholder users a Direct Transfer creates. "exclude_internal" alone keeps
    # project bots and "active=true" drops placeholders, so the old arithmetic
    # mislabelled them as humans resp. deactivated users.
    humans = _count({"humans": "true"})
    if humans is not None and humans >= total:
        # Older GitLab ignores unknown filters - fall back to the coarser one.
        humans = _count({"exclude_internal": "true"})
        human_filter = {"exclude_internal": "true"}
    else:
        human_filter = {"humans": "true"}
    active = _count({"active": "true"})
    billable = _count({**human_filter, "active": "true"})
    blocked = _count({"blocked": "true"})
    blocked_humans = _count({**human_filter, "blocked": "true"})

    bots = (total - humans) if (humans is not None) else None
    deactivated = None
    if humans is not None and billable is not None and blocked_humans is not None:
        # Deactivated humans: neither active nor blocked.
        deactivated = max(0, humans - billable - blocked_humans)

    return {
        "count_unavailable": False,
        "total": total,
        "humans": humans,
        "bots": bots,              # every non-human account (bots, placeholders, internal)
        "active": active,          # active (any account, incl. active bots)
        "blocked": blocked,
        "deactivated": deactivated,  # human accounts that are neither active nor blocked
        "billable": billable,      # active human accounts == 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 collect_status_page(args: Args) -> dict[str, Any]:
    """Public GitLab.com status page (status.io API, no authentication).

    Reduced to overall status, components (with their containers/regions),
    active incidents and active maintenances. See
    https://kb.status.io/developers/public-status-api/
    """
    url = STATUS_IO_API.format(page_id=urllib.parse.quote(args.status_page_id))
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    try:
        with urllib.request.urlopen(request, timeout=args.timeout) as resp:
            data = json.loads(resp.read())
    except (urllib.error.URLError, OSError, ValueError) as exc:
        if args.debug:
            raise
        return {"error": f"{url}: {exc}"}

    result = data.get("result") if isinstance(data, dict) else None
    if not isinstance(result, dict):
        return {"error": f"{url}: unexpected response"}

    def _status(obj: dict[str, Any]) -> dict[str, Any]:
        return {"name": obj.get("name"), "status": obj.get("status"),
                "status_code": obj.get("status_code")}

    components = []
    for comp in result.get("status") or []:
        if not isinstance(comp, dict):
            continue
        entry = _status(comp)
        entry["containers"] = [_status(c) for c in comp.get("containers") or []
                               if isinstance(c, dict)]
        components.append(entry)

    maintenance = result.get("maintenance") or {}
    return {
        "url": "https://status.gitlab.com/",
        "history_url": f"https://status.gitlab.com/pages/history/{args.status_page_id}",
        "overall": _status(result.get("status_overall") or {}),
        "updated": (result.get("status_overall") or {}).get("updated"),
        "components": components,
        "incidents": [{"name": i.get("name")} for i in result.get("incidents") or []
                      if isinstance(i, dict)],
        "maintenance": [{"name": m.get("name")} for m in maintenance.get("active") or []
                        if isinstance(m, dict)],
    }


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_mirrors:
        try:
            mirrors = collect_mirrors(api, args, [p for p, _ref, _cj in targets])
        except RuntimeError as exc:
            if args.debug:
                raise
            mirrors = []
            sys.stderr.write(f"Mirror collection failed: {exc}\n")
        _emit("gitlab_mirrors", mirrors)

    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 {})

    if args.collect_status_page:
        _emit("gitlab_status", collect_status_page(args))

    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())
