"""Deployment plan and host preflight checks."""

from __future__ import annotations

from dataclasses import asdict, dataclass
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import socket
import ssl
import stat
import subprocess
import time
from typing import Callable, Mapping, Sequence
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

from .backup_evidence import (
    MAX_BACKUP_EVIDENCE_BYTES,
    MAX_BACKUP_KEYRING_BYTES,
    verify_backup_evidence,
)
from .bundle import (
    BundlePaths,
    canonical_json,
    digest_json,
    environment_fingerprint,
    read_env,
    render_caddy_config,
    render_compose,
    render_existing_proxy_contract,
    service_names,
)
from .distribution import (
    MAX_KEYRING_BYTES,
    MAX_MANIFEST_BYTES,
    DistributionError,
    canonical_json as canonical_distribution_json,
    decode_json_bytes,
    file_sha256,
    load_bounded_json,
    read_bounded_bytes,
    verify_manifest,
    verify_manifest_binding,
)
from .model import (
    InstallationSpec,
    image_is_digest_pinned,
    image_is_unpublished,
)


@dataclass(frozen=True, slots=True)
class Check:
    id: str
    level: str
    message: str
    action: str = ""

    def to_dict(self) -> dict[str, str]:
        return asdict(self)


@dataclass(frozen=True, slots=True)
class PlanAction:
    action: str
    target: str
    detail: str

    def to_dict(self) -> dict[str, str]:
        return asdict(self)


@dataclass(frozen=True, slots=True)
class DeploymentPlan:
    installation_id: str
    desired_spec_sha256: str
    desired_compose_sha256: str
    desired_environment_fingerprint: str
    actions: tuple[PlanAction, ...]
    checks: tuple[Check, ...]

    @property
    def blocked(self) -> bool:
        return any(check.level == "error" for check in self.checks)

    def to_dict(self) -> dict[str, object]:
        return {
            "installation_id": self.installation_id,
            "desired_spec_sha256": self.desired_spec_sha256,
            "desired_compose_sha256": self.desired_compose_sha256,
            "desired_environment_fingerprint": (self.desired_environment_fingerprint),
            "blocked": self.blocked,
            "actions": [action.to_dict() for action in self.actions],
            "checks": [check.to_dict() for check in self.checks],
        }


CommandRunner = Callable[[Sequence[str], Path], subprocess.CompletedProcess[str]]


def build_plan(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    include_host_checks: bool = True,
    command_runner: CommandRunner | None = None,
) -> DeploymentPlan:
    compose = render_compose(spec)
    desired_services = set(service_names(spec))
    previous = _read_receipt(paths.receipt)
    previous_services = {
        str(item) for item in previous.get("services", []) if isinstance(item, str)
    }
    previous_spec_digest = previous.get("spec_sha256")
    previous_compose_digest = previous.get("compose_sha256")
    previous_environment_fingerprint = previous.get("environment_fingerprint")
    spec_digest = digest_json(spec.to_dict())
    compose_digest = digest_json(compose)
    environment_digest = environment_fingerprint(read_env(paths.env))

    actions: list[PlanAction] = []
    if not previous:
        actions.append(
            PlanAction(
                "create",
                "installation",
                "Create the first deployment revision.",
            )
        )
    if previous_spec_digest != spec_digest:
        actions.append(
            PlanAction(
                "reconfigure",
                "installation",
                "Reconcile runtime configuration with installation.json.",
            )
        )
    if previous_compose_digest != compose_digest:
        actions.append(
            PlanAction(
                "render",
                "compose",
                "Render a new deterministic Compose definition.",
            )
        )
    if previous_environment_fingerprint != environment_digest:
        actions.append(
            PlanAction(
                "reconfigure",
                "environment",
                "Reconcile changed secret bindings or runtime environment values.",
            )
        )
    for name in sorted(desired_services - previous_services):
        actions.append(PlanAction("start", name, "Start the selected service."))
    for name in sorted(previous_services - desired_services):
        actions.append(
            PlanAction(
                "remove",
                name,
                "Remove the service container; retained volumes are not deleted.",
            )
        )
    if (
        previous
        and previous_spec_digest == spec_digest
        and previous_compose_digest == compose_digest
        and previous_environment_fingerprint == environment_digest
        and previous_services == desired_services
    ):
        actions.append(
            PlanAction(
                "noop", "installation", "Desired state matches the last receipt."
            )
        )

    checks = list(static_checks(spec, paths))
    if include_host_checks:
        checks.extend(host_checks(spec, paths, command_runner=command_runner))
    return DeploymentPlan(
        installation_id=spec.installation_id,
        desired_spec_sha256=spec_digest,
        desired_compose_sha256=compose_digest,
        desired_environment_fingerprint=environment_digest,
        actions=tuple(actions),
        checks=tuple(checks),
    )


def static_checks(spec: InstallationSpec, paths: BundlePaths) -> tuple[Check, ...]:
    checks: list[Check] = []
    checks.extend(_ingress_configuration_checks(spec, paths))
    images = {
        "release.api_image": spec.release.api_image,
        "release.web_image": spec.release.web_image,
    }
    if spec.components.postgres.mode == "managed":
        images["components.postgres.image"] = spec.components.postgres.image
    if spec.components.redis.mode == "managed":
        images["components.redis.image"] = spec.components.redis.image
    if spec.components.mail.mode == "test-mail":
        images["components.mail.image"] = spec.components.mail.image
    if spec.components.storage.mode == "garage":
        images["components.storage.image"] = spec.components.storage.image
    images["components.load_balancer.image"] = spec.components.load_balancer.image
    if spec.ingress.mode == "managed":
        images["ingress.image"] = spec.ingress.image

    for label, image in images.items():
        if image_is_unpublished(image):
            checks.append(
                Check(
                    f"{label}.published",
                    "error",
                    f"{label} still uses the unpublished placeholder.",
                    "Set an available image reference before apply.",
                )
            )
        elif not image_is_digest_pinned(image):
            level = "error" if spec.profile == "self-hosted" else "warning"
            checks.append(
                Check(
                    f"{label}.digest",
                    level,
                    f"{label} is not pinned by OCI digest.",
                    "Use an image@sha256:... reference from a verified distribution.",
                )
            )
        else:
            checks.append(
                Check(
                    f"{label}.digest",
                    "ok",
                    f"{label} is pinned by OCI digest.",
                )
            )

    checks.extend(_distribution_checks(spec, paths))
    checks.extend(
        _backup_evidence_checks(spec, paths, receipt=_read_receipt(paths.receipt))
    )

    values = read_env(paths.env)
    required = {"MASTER_KEY_B64", "DATABASE_URL"}
    if spec.components.redis.mode != "disabled":
        required.add("REDIS_URL")
    if spec.components.storage.mode in {"s3", "garage"}:
        required.update(
            {
                "FILE_STORAGE_S3_ENDPOINT_URL",
                "FILE_STORAGE_S3_REGION",
                "FILE_STORAGE_S3_ACCESS_KEY_ID",
                "FILE_STORAGE_S3_SECRET_ACCESS_KEY",
                "FILE_STORAGE_S3_BUCKET",
            }
        )
    if spec.components.storage.mode == "garage":
        required.update(
            {
                "GARAGE_DEFAULT_ACCESS_KEY",
                "GARAGE_DEFAULT_SECRET_KEY",
                "GARAGE_DEFAULT_BUCKET",
                "GARAGE_RPC_SECRET",
                "GARAGE_ADMIN_TOKEN",
                "GARAGE_METRICS_TOKEN",
            }
        )
    missing = sorted(name for name in required if not values.get(name))
    checks.append(
        Check(
            "secrets.required",
            "error" if missing else "ok",
            (
                "Missing required secret values: " + ", ".join(missing)
                if missing
                else "Required runtime secret references are populated."
            ),
            "Re-run configure with the required external service values."
            if missing
            else "",
        )
    )
    if paths.env.exists():
        mode = stat.S_IMODE(paths.env.stat().st_mode)
        checks.append(
            Check(
                "secrets.permissions",
                "error" if mode & 0o077 else "ok",
                (
                    f"{paths.env.name} has unsafe mode {mode:04o}."
                    if mode & 0o077
                    else f"{paths.env.name} is private ({mode:04o})."
                ),
                f"Run chmod 600 {paths.env}" if mode & 0o077 else "",
            )
        )
    if spec.profile == "self-hosted":
        checks.append(
            Check(
                "bootstrap.administrator",
                "warning",
                "Production first-administrator enrollment is not automated yet.",
                "Use the controlled one-time administrator procedure until the enrollment slice lands.",
            )
        )
    if spec.components.mail.mode == "external-relay":
        checks.append(
            Check(
                "mail.provisioning",
                "warning",
                "The external relay is selected but Mail profile seeding remains an explicit post-install configuration task.",
                "Create the reusable server and credential profile in Mail after first login.",
            )
        )
    if spec.components.storage.mode == "local" and spec.profile == "self-hosted":
        checks.append(
            Check(
                "storage.local",
                "warning",
                "Local file storage is suitable for one host but prevents stateless API scale-out.",
                "Include files-data in backup/restore drills or configure S3 storage.",
            )
        )
    if spec.components.storage.mode == "garage":
        checks.append(
            Check(
                "storage.garage.single_node",
                "warning",
                (
                    "Managed Garage is persistent S3-compatible storage, but "
                    "this Compose profile runs one Garage node without data redundancy."
                ),
                (
                    "Use an external multi-node Garage/S3 service and tested "
                    "backup/restore for an availability-sensitive installation."
                ),
            )
        )
    checks.append(
        Check(
            "topology.load_balancing",
            "ok",
            (
                "Managed HAProxy balances "
                f"{spec.replicas.web} WebUI and {spec.replicas.api} API replica(s)."
            ),
        )
    )
    if spec.replicas.worker > 1:
        checks.append(
            Check(
                "topology.worker_scaling",
                "ok",
                f"{spec.replicas.worker} workers share the configured Redis queues.",
            )
        )
    return tuple(checks)


def release_change_requires_backup(
    spec: InstallationSpec,
    receipt: Mapping[str, object],
) -> bool:
    if spec.profile != "self-hosted" or not receipt:
        return False
    previous = receipt.get("release")
    if not isinstance(previous, Mapping):
        return True
    desired = {
        "channel": spec.release.channel,
        "version": spec.release.version,
        "manifest_sha256": spec.release.manifest_sha256,
        "composition_sha256": spec.release.composition_sha256,
        "api_image": spec.release.api_image,
        "web_image": spec.release.web_image,
    }
    return any(previous.get(key) != value for key, value in desired.items())


def verify_stored_backup_evidence(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    receipt: Mapping[str, object],
) -> dict[str, object]:
    verification = load_bounded_json(
        paths.backup_verification,
        maximum_bytes=64 * 1024,
    )
    expected_fields = {
        "schema_version",
        "evidence_sha256",
        "keyring_sha256",
        "signature_key_id",
        "verified_at",
        "evidence_id",
        "recovery_point_id",
        "restore_drill_id",
        "release_manifest_sha256",
        "captured_at",
        "expires_at",
        "restore_started_at",
        "restore_completed_at",
        "measured_rpo_seconds",
        "measured_rto_seconds",
        "component_count",
    }
    if set(verification) != expected_fields or verification.get("schema_version") != 1:
        raise DistributionError("backup verification receipt is malformed")
    encoded_evidence = read_bounded_bytes(
        paths.backup_evidence,
        maximum_bytes=MAX_BACKUP_EVIDENCE_BYTES,
    )
    encoded_keyring = read_bounded_bytes(
        paths.backup_keyring,
        maximum_bytes=MAX_BACKUP_KEYRING_BYTES,
    )
    evidence = decode_json_bytes(encoded_evidence, label="backup evidence")
    keyring = decode_json_bytes(encoded_keyring, label="backup keyring")
    if encoded_evidence != canonical_distribution_json(evidence):
        raise DistributionError("stored backup evidence is not canonical JSON")
    if encoded_keyring != canonical_distribution_json(keyring):
        raise DistributionError("stored backup keyring is not canonical JSON")
    evidence_digest = hashlib.sha256(encoded_evidence).hexdigest()
    keyring_digest = hashlib.sha256(encoded_keyring).hexdigest()
    if evidence_digest != verification.get("evidence_sha256"):
        raise DistributionError("stored backup evidence digest has changed")
    if keyring_digest != verification.get("keyring_sha256"):
        raise DistributionError("stored backup keyring digest has changed")
    previous_release = receipt.get("release") if receipt else None
    expected_release: Mapping[str, object] = (
        previous_release
        if isinstance(previous_release, Mapping)
        else {
            "channel": spec.release.channel,
            "version": spec.release.version,
            "manifest_sha256": spec.release.manifest_sha256,
            "composition_sha256": spec.release.composition_sha256,
            "api_image": spec.release.api_image,
            "web_image": spec.release.web_image,
        }
    )
    summary = verify_backup_evidence(
        evidence,
        keyring,
        installation_id=spec.installation_id,
        profile=spec.profile,
        release=expected_release,
    )
    expected_summary = {
        "signature_key_id": verification.get("signature_key_id"),
        "evidence_id": verification.get("evidence_id"),
        "recovery_point_id": verification.get("recovery_point_id"),
        "restore_drill_id": verification.get("restore_drill_id"),
        "captured_at": verification.get("captured_at"),
        "expires_at": verification.get("expires_at"),
        "restore_started_at": verification.get("restore_started_at"),
        "restore_completed_at": verification.get("restore_completed_at"),
        "measured_rpo_seconds": verification.get("measured_rpo_seconds"),
        "measured_rto_seconds": verification.get("measured_rto_seconds"),
        "component_count": verification.get("component_count"),
    }
    for field, expected in expected_summary.items():
        if summary.get(field) != expected:
            raise DistributionError(
                f"backup verification receipt does not match {field!r}"
            )
    if expected_release.get("manifest_sha256") != verification.get(
        "release_manifest_sha256"
    ):
        raise DistributionError("backup verification receipt has another release")
    return {
        **summary,
        "evidence_sha256": evidence_digest,
        "keyring_sha256": keyring_digest,
    }


def _backup_evidence_checks(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    receipt: Mapping[str, object],
) -> tuple[Check, ...]:
    required = release_change_requires_backup(spec, receipt)
    available = all(
        path.is_file()
        for path in (
            paths.backup_evidence,
            paths.backup_keyring,
            paths.backup_verification,
        )
    )
    if not available:
        return (
            Check(
                "backup.migration_gate",
                "error"
                if required
                else "warning"
                if spec.profile == "self-hosted"
                else "ok",
                (
                    "A release-changing migration has no verified coordinated backup evidence."
                    if required
                    else "No current coordinated backup evidence is adopted."
                ),
                (
                    "Run verify-backup --adopt after an isolated restore drill."
                    if spec.profile == "self-hosted"
                    else ""
                ),
            ),
        )
    try:
        summary = verify_stored_backup_evidence(
            spec,
            paths,
            receipt=receipt,
        )
    except (DistributionError, OSError) as exc:
        return (
            Check(
                "backup.migration_gate",
                "error" if required else "warning",
                f"Coordinated backup evidence is invalid: {exc}",
                "Adopt fresh signed evidence for the currently applied release.",
            ),
        )
    return (
        Check(
            "backup.migration_gate",
            "ok",
            (
                "Release migration is backed by recovery point "
                f"{summary['recovery_point_id']} and restore drill "
                f"{summary['restore_drill_id']}."
            ),
        ),
    )


def _ingress_configuration_checks(
    spec: InstallationSpec,
    paths: BundlePaths,
) -> tuple[Check, ...]:
    if spec.ingress.mode == "unconfigured":
        return (
            Check(
                "ingress.configuration",
                "error" if spec.profile == "self-hosted" else "warning",
                "No supported public ingress boundary is configured.",
                "Select managed ingress or an existing reverse proxy before apply.",
            ),
        )
    checks = [
        Check(
            "ingress.configuration",
            "ok",
            f"Ingress mode {spec.ingress.mode!r} has a bounded configuration.",
        )
    ]
    if spec.ingress.mode == "managed":
        checks.append(
            _artifact_check(
                "ingress.managed_config",
                paths.caddy_config,
                render_caddy_config(spec).encode("utf-8"),
                "Managed ingress configuration",
            )
        )
        ingress = render_compose(spec)["services"].get("ingress", {})
        volumes = ingress.get("volumes", []) if isinstance(ingress, dict) else []
        persistent = "caddy-data:/data" in volumes and "caddy-config:/config" in volumes
        checks.append(
            Check(
                "ingress.certificate_state",
                "ok" if persistent else "error",
                (
                    "Managed certificate and renewal state uses persistent private volumes."
                    if persistent
                    else "Managed certificate state is not persistent."
                ),
                "Restore the caddy-data and caddy-config volume bindings."
                if not persistent
                else "",
            )
        )
    elif spec.ingress.mode == "existing-proxy":
        checks.append(
            _artifact_check(
                "ingress.existing_proxy_contract",
                paths.existing_proxy,
                canonical_json(render_existing_proxy_contract(spec)),
                "Existing reverse-proxy contract",
            )
        )
    return tuple(checks)


def _artifact_check(
    check_id: str,
    path: Path,
    expected: bytes,
    label: str,
) -> Check:
    try:
        actual = path.read_bytes()
    except OSError as exc:
        return Check(
            check_id,
            "error",
            f"{label} is unavailable: {exc}",
            "Re-render the installation bundle.",
        )
    matches = actual == expected
    return Check(
        check_id,
        "ok" if matches else "error",
        f"{label} {'matches' if matches else 'does not match'} the installation specification.",
        "Re-render the installation bundle before apply." if not matches else "",
    )


def _distribution_checks(
    spec: InstallationSpec,
    paths: BundlePaths,
) -> tuple[Check, ...]:
    blocking_level = "error" if spec.profile == "self-hosted" else "warning"
    if not (
        spec.release.manifest_sha256
        and spec.release.manifest_keyring_sha256
        and spec.release.manifest_signature_key_id
        and spec.release.composition_sha256
        and paths.manifest.exists()
        and paths.keyring.exists()
    ):
        return (
            Check(
                "release.manifest",
                blocking_level,
                "No locally verified runtime distribution is recorded.",
                "Run govoplan-deploy verify-release --adopt with an independently trusted keyring.",
            ),
            Check(
                "release.signature_verification",
                blocking_level,
                "Runtime distribution signature evidence is unavailable.",
                "Install and verify the signed distribution before apply.",
            ),
            Check(
                "modules.image_composition",
                blocking_level,
                "Enabled modules are not bound to image composition evidence.",
                "Adopt a distribution whose composition contains every enabled module.",
            ),
        )
    try:
        manifest_digest = file_sha256(
            paths.manifest,
            maximum_bytes=MAX_MANIFEST_BYTES,
        )
        if manifest_digest != spec.release.manifest_sha256:
            raise DistributionError(
                "stored manifest digest does not match installation"
            )
        keyring_digest = file_sha256(
            paths.keyring,
            maximum_bytes=MAX_KEYRING_BYTES,
        )
        if keyring_digest != spec.release.manifest_keyring_sha256:
            raise DistributionError("stored keyring digest does not match installation")
        manifest = load_bounded_json(
            paths.manifest,
            maximum_bytes=MAX_MANIFEST_BYTES,
        )
        keyring = load_bounded_json(
            paths.keyring,
            maximum_bytes=MAX_KEYRING_BYTES,
        )
        key_id = verify_manifest(
            manifest,
            keyring,
            expected_channel=spec.release.channel,
        )
        if key_id != spec.release.manifest_signature_key_id:
            raise DistributionError(
                "verified signature key does not match installation"
            )
        verify_manifest_binding(
            manifest,
            channel=spec.release.channel,
            version=spec.release.version,
            api_image=spec.release.api_image,
            web_image=spec.release.web_image,
            enabled_modules=spec.enabled_modules,
            composition_sha256=spec.release.composition_sha256,
            dependencies=_selected_dependency_images(spec),
        )
    except (DistributionError, OSError) as exc:
        return (
            Check(
                "release.manifest",
                blocking_level,
                f"Runtime distribution verification failed: {exc}",
                "Re-adopt an unexpired, non-revoked manifest from a trusted release key.",
            ),
            Check(
                "release.signature_verification",
                blocking_level,
                "Runtime distribution signature is not trusted.",
                "Correct the manifest/keyring binding before apply.",
            ),
            Check(
                "modules.image_composition",
                blocking_level,
                "Runtime image composition is not trusted.",
                "Correct the signed composition binding before apply.",
            ),
        )
    return (
        Check(
            "release.manifest",
            "ok",
            "Stored runtime distribution matches its independently pinned digest.",
        ),
        Check(
            "release.signature_verification",
            "ok",
            f"Runtime distribution is signed by trusted key {key_id}.",
        ),
        Check(
            "modules.image_composition",
            "ok",
            "Every enabled module is present in signed image composition evidence.",
        ),
    )


def _selected_dependency_images(spec: InstallationSpec) -> dict[str, str]:
    values = {"load_balancer": spec.components.load_balancer.image}
    if spec.components.postgres.mode == "managed":
        values["postgres"] = spec.components.postgres.image
    if spec.components.redis.mode == "managed":
        values["redis"] = spec.components.redis.image
    if spec.components.mail.mode == "test-mail":
        values["test_mail"] = spec.components.mail.image
    if spec.components.storage.mode == "garage":
        values["garage"] = spec.components.storage.image
    if spec.ingress.mode == "managed":
        values["managed_ingress"] = spec.ingress.image
    return values


def host_checks(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    command_runner: CommandRunner | None = None,
) -> tuple[Check, ...]:
    checks: list[Check] = []
    runner = command_runner or _run_command
    machine = platform.machine().lower()
    supported = machine in {"x86_64", "amd64", "aarch64", "arm64"}
    checks.append(
        Check(
            "host.architecture",
            "ok" if supported else "error",
            f"Host architecture is {machine or 'unknown'}.",
            "Use a supported amd64 or arm64 host." if not supported else "",
        )
    )

    memory_bytes = _memory_bytes()
    if memory_bytes is not None:
        minimum = 4 * 1024**3
        checks.append(
            Check(
                "host.memory",
                "ok" if memory_bytes >= minimum else "warning",
                f"Host memory is approximately {memory_bytes / 1024**3:.1f} GiB.",
                "Use at least 4 GiB for the base container profile."
                if memory_bytes < minimum
                else "",
            )
        )
    cpu_count = os.cpu_count()
    if cpu_count is not None:
        checks.append(
            Check(
                "host.cpu",
                "ok" if cpu_count >= 2 else "warning",
                f"Host reports {cpu_count} logical CPU(s).",
                "Use at least two logical CPUs for the base container profile."
                if cpu_count < 2
                else "",
            )
        )
    entropy = _entropy_available()
    if entropy is not None:
        checks.append(
            Check(
                "host.entropy",
                "ok" if entropy >= 256 else "warning",
                f"Kernel entropy availability is {entropy}.",
                "Wait for or provide sufficient host entropy before generating production keys."
                if entropy < 256
                else "",
            )
        )
    free_bytes = shutil.disk_usage(paths.root).free
    minimum_free = 10 * 1024**3
    checks.append(
        Check(
            "host.disk",
            "ok" if free_bytes >= minimum_free else "warning",
            f"Free installation filesystem space is approximately {free_bytes / 1024**3:.1f} GiB.",
            "Keep at least 10 GiB free before pulling images and creating data volumes."
            if free_bytes < minimum_free
            else "",
        )
    )

    docker = shutil.which("docker")
    if docker is None:
        checks.append(
            Check(
                "host.compose",
                "error",
                "Docker CLI is not installed or not on PATH.",
                "Install Docker Engine with the Compose v2 plugin.",
            )
        )
    else:
        result = runner((docker, "compose", "version", "--short"), paths.root)
        checks.append(
            Check(
                "host.compose",
                "ok" if result.returncode == 0 else "error",
                (
                    f"Docker Compose is available ({result.stdout.strip()})."
                    if result.returncode == 0
                    else "Docker Compose v2 is not available."
                ),
                "Install or enable the Docker Compose v2 plugin."
                if result.returncode != 0
                else "",
            )
        )
        daemon = runner(
            (docker, "info", "--format", "{{.ServerVersion}}"),
            paths.root,
        )
        checks.append(
            Check(
                "host.container_runtime",
                "ok" if daemon.returncode == 0 else "error",
                (
                    f"Docker daemon is reachable ({daemon.stdout.strip()})."
                    if daemon.returncode == 0
                    else "Docker CLI cannot reach the Docker daemon."
                ),
                "Start Docker and grant this operator access to the daemon."
                if daemon.returncode != 0
                else "",
            )
        )

    values = read_env(paths.env)
    if spec.components.postgres.mode == "external":
        checks.append(
            _endpoint_check(
                "external.postgres",
                "External PostgreSQL",
                values.get("DATABASE_URL", ""),
                default_ports={"postgresql": 5432, "postgresql+psycopg": 5432},
            )
        )
    if spec.components.redis.mode == "external":
        checks.append(
            _endpoint_check(
                "external.redis",
                "External Redis",
                values.get("REDIS_URL", ""),
                default_ports={"redis": 6379, "rediss": 6379},
            )
        )
    if spec.components.storage.mode == "s3":
        checks.append(
            _endpoint_check(
                "external.s3",
                "S3-compatible storage",
                values.get("FILE_STORAGE_S3_ENDPOINT_URL", ""),
                default_ports={"http": 80, "https": 443},
            )
        )

    receipt = _read_receipt(paths.receipt)
    checks.extend(
        _ingress_host_checks(
            spec,
            paths,
            receipt=receipt,
            docker=docker,
            command_runner=runner,
        )
    )
    return tuple(checks)


def _ingress_host_checks(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    receipt: Mapping[str, object],
    docker: str | None,
    command_runner: CommandRunner,
) -> tuple[Check, ...]:
    checks: list[Check] = []
    applied = bool(receipt)
    if spec.ingress.mode in {"managed", "existing-proxy"}:
        public = urlsplit(spec.public_url)
        host = public.hostname or ""
        port = public.port or 443
        checks.append(_dns_resolution_check(host, port))
        if spec.ingress.mode == "managed" and not applied:
            checks.append(
                Check(
                    "ingress.tls",
                    "warning",
                    "TLS issuance will be verified after managed ingress starts.",
                    "Ensure public DNS resolves to this host and ports 80/443 are reachable.",
                )
            )
            checks.append(
                Check(
                    "ingress.public_route",
                    "warning",
                    "Public-route health will be verified after managed ingress starts.",
                    "Permit inbound HTTP and HTTPS through the host firewall and upstream NAT.",
                )
            )
        else:
            checks.append(_tls_validity_check(host, port))
            checks.append(
                _public_route_check(
                    spec.public_url,
                    require_ready=applied,
                )
            )

    previous_ingress = receipt.get("ingress")
    desired_ingress = {
        "mode": spec.ingress.mode,
        "http_port": spec.ingress.http_port,
        "https_port": spec.ingress.https_port,
    }
    previous_listen = receipt.get("listen")
    desired_listen = {
        "address": spec.listen.address,
        "port": spec.listen.port,
    }
    if spec.ingress.mode == "managed":
        if not applied or previous_ingress != desired_ingress:
            for label, port in (
                ("http", spec.ingress.http_port),
                ("https", spec.ingress.https_port),
            ):
                checks.append(
                    _available_port_check(
                        f"host.ingress_{label}_port",
                        "0.0.0.0",
                        port,
                    )
                )
    elif not applied or previous_listen != desired_listen:
        checks.append(
            _available_port_check(
                "host.listen_port",
                spec.listen.address,
                spec.listen.port,
            )
        )

    if applied:
        checks.append(
            _local_upstream_check(
                spec,
                paths,
                docker=docker,
                command_runner=command_runner,
            )
        )
    return tuple(checks)


def _available_port_check(check_id: str, address: str, port: int) -> Check:
    available = _port_available(address, port)
    return Check(
        check_id,
        "ok" if available else "error",
        (
            f"Listen endpoint {address}:{port} is available."
            if available
            else f"Listen endpoint {address}:{port} is already in use."
        ),
        "Choose another port or stop the conflicting service." if not available else "",
    )


def _dns_resolution_check(host: str, port: int) -> Check:
    try:
        results = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
        addresses = sorted({str(item[4][0]) for item in results})
    except OSError as exc:
        return Check(
            "ingress.dns",
            "error",
            f"Public hostname {host!r} does not resolve: {exc}",
            "Publish public A/AAAA records before apply.",
        )
    return Check(
        "ingress.dns",
        "ok",
        f"Public hostname {host!r} resolves to {', '.join(addresses[:8])}.",
    )


def _tls_validity_check(host: str, port: int) -> Check:
    try:
        context = ssl.create_default_context()
        with socket.create_connection((host, port), timeout=3.0) as connection:
            with context.wrap_socket(connection, server_hostname=host) as secured:
                certificate = secured.getpeercert()
        expires = str(certificate.get("notAfter") or "")
        remaining_seconds = ssl.cert_time_to_seconds(expires) - time.time()
    except (OSError, ValueError, ssl.SSLError) as exc:
        return Check(
            "ingress.tls",
            "error",
            f"Public TLS validation failed for {host}:{port}: {exc}",
            "Correct certificate issuance, trust chain, hostname, and public routing.",
        )
    remaining_days = int(remaining_seconds // 86400)
    level = "ok" if remaining_days >= 21 else "warning"
    return Check(
        "ingress.tls",
        level,
        f"Public TLS certificate is valid for approximately {remaining_days} more day(s).",
        "Verify automated certificate renewal immediately."
        if level == "warning"
        else "",
    )


def _public_route_check(public_url: str, *, require_ready: bool) -> Check:
    target = public_url.rstrip("/") + "/health/ready"
    status: int | None = None
    try:
        request = Request(target, headers={"User-Agent": "govoplan-deploy/doctor"})
        with urlopen(request, timeout=4.0) as response:
            status = response.status
    except HTTPError as exc:
        status = exc.code
    except (OSError, URLError, ValueError) as exc:
        return Check(
            "ingress.public_route",
            "error",
            f"Public route is unreachable: {exc}",
            "Check external DNS, firewall/NAT, reverse-proxy routing, and TLS.",
        )
    acceptable = status == 200 if require_ready else status not in {400, 404, 421}
    return Check(
        "ingress.public_route",
        "ok" if acceptable else "error",
        f"Public readiness route returned HTTP {status}.",
        (
            "Route the configured hostname and /health/ready path to the generated upstream."
            if not acceptable
            else ""
        ),
    )


def _local_upstream_check(
    spec: InstallationSpec,
    paths: BundlePaths,
    *,
    docker: str | None,
    command_runner: CommandRunner,
) -> Check:
    if docker is None:
        return Check(
            "ingress.local_upstream",
            "error",
            "Local upstream health cannot be checked without Docker.",
            "Restore Docker access and rerun doctor.",
        )
    argv = (
        docker,
        "compose",
        "--env-file",
        str(paths.env),
        "--project-name",
        spec.installation_id,
        "--file",
        str(paths.compose),
        "exec",
        "--no-TTY",
        "load-balancer",
        "wget",
        "-qO-",
        "http://127.0.0.1:8080/health",
    )
    try:
        result = command_runner(argv, paths.root)
    except (OSError, subprocess.SubprocessError) as exc:
        return Check(
            "ingress.local_upstream",
            "error",
            f"Local upstream health probe failed: {exc}",
            "Inspect the load-balancer and WebUI service health.",
        )
    return Check(
        "ingress.local_upstream",
        "ok" if result.returncode == 0 else "error",
        (
            "The generated local upstream is healthy."
            if result.returncode == 0
            else "The generated local upstream health probe failed."
        ),
        "Inspect load-balancer and WebUI health before exposing the route."
        if result.returncode != 0
        else "",
    )


def _read_receipt(path: Path) -> Mapping[str, object]:
    if not path.exists():
        return {}
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return value if isinstance(value, dict) else {}


def _memory_bytes() -> int | None:
    try:
        for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
            if line.startswith("MemTotal:"):
                return int(line.split()[1]) * 1024
    except (OSError, ValueError, IndexError):
        return None
    return None


def _entropy_available() -> int | None:
    try:
        return int(
            Path("/proc/sys/kernel/random/entropy_avail")
            .read_text(encoding="utf-8")
            .strip()
        )
    except (OSError, ValueError):
        return None


def _port_available(address: str, port: int) -> bool:
    family = socket.AF_INET6 if ":" in address else socket.AF_INET
    with socket.socket(family, socket.SOCK_STREAM) as handle:
        try:
            handle.bind((address, port))
        except OSError:
            return False
    return True


def _endpoint_check(
    check_id: str,
    label: str,
    url: str,
    *,
    default_ports: Mapping[str, int],
) -> Check:
    try:
        parsed = urlsplit(url)
        host = parsed.hostname
        port = parsed.port or default_ports.get(parsed.scheme)
    except ValueError as exc:
        return Check(
            check_id,
            "error",
            f"{label} endpoint is invalid: {exc}",
            "Correct the private endpoint binding and rerun doctor.",
        )
    if not host or port is None:
        return Check(
            check_id,
            "error",
            f"{label} endpoint has no usable host and port.",
            "Correct the private endpoint binding and rerun doctor.",
        )
    try:
        with socket.create_connection((host, port), timeout=1.5):
            pass
    except OSError as exc:
        return Check(
            check_id,
            "error",
            f"{label} is not reachable at {host}:{port}: {exc}",
            "Check DNS, routing, firewall, service health, and the selected endpoint.",
        )
    return Check(
        check_id,
        "ok",
        f"{label} is reachable at {host}:{port}.",
    )


def _run_command(argv: Sequence[str], cwd: Path) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        list(argv),
        cwd=cwd,
        check=False,
        capture_output=True,
        text=True,
        timeout=10,
    )