"""CI/CD use cases: the delivery hooks, the pipeline executor and the demo runtime.

This module owns the whole "push to `dev` becomes a live demo" story (D7). Three
decisions shape everything below.

**A failed run never takes the demo down.** Blue/green is not an optimisation
here, it is the safety property. The replacement container is started on a
*staging* port, healthchecked, and only then does the Caddy route move. If any
step fails, the old container keeps running untouched and the run is marked
failed with a plain-language reason. Deploying the wrong code silently is the
agybot bug this design exists to prevent.

**Logs stream; only a tail is stored.** Each step's output is pushed as batched
realtime events, and a bounded tail is persisted for the red case. A build log is
unbounded and a database is the wrong place for it — but the last few KB is
exactly what someone needs when a step fails.

**Nothing here reaches production.** That is a separate, human-triggered path
(Phase 5), and there is deliberately no code path from a pipeline run into it.

The demo port ledger is the *database*, not the router adapter's in-memory map:
the API and the delivery worker are different processes, and an in-process port
map would hand the same port to two projects the moment a second process started.
"""

from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Protocol, runtime_checkable

from app.application.ports.domain_router import DomainRouterPort
from app.application.ports.realtime import RealtimeEvent, get_publisher, org_channel
from app.application.ports.repo_workspace import RepoWorkspacePort
from app.core.config import settings
from app.core.errors import ConflictError, NotFoundError, ValidationError
from app.core.redact import redact_text
from app.domain.delivery.models import (
    OPTIONAL_STEPS,
    PIPELINE_STEPS,
    AppRuntime,
    PipelineRun,
    PipelineStep,
    Release,
    RunStatus,
    RuntimeStatus,
    RunTrigger,
    StepStatus,
)
from app.domain.platform.rules import (
    RuleViolation,
    assert_port_allocatable,
    build_host,
)
from app.domain.projects.profile import AppProfile
from app.infrastructure.adapters.docker_sandbox import (
    FALLBACK_IMAGE,
    ROLE_BUILD,
    ROLE_DEMO,
    image_for_language,
)
from app.infrastructure.adapters.process import ProcessResult
from app.infrastructure.repositories.app_profile import AppProfileRepository
from app.infrastructure.repositories.delivery import (
    AppRuntimeRepository,
    PipelineRunRepository,
    PipelineStepRepository,
    ReleaseRepository,
)
from app.infrastructure.repositories.identity import OrganizationRepository
from app.infrastructure.repositories.projects import (
    PreviewDomainRepository,
    ProjectRepository,
)

logger = logging.getLogger("agy.delivery")

#: Per-step timeout. A build gets longer than a test run; neither is unbounded.
STEP_TIMEOUTS: dict[str, float] = {
    "checkout": 180.0,
    "setup": 900.0,
    "build": 900.0,
    "test": 600.0,
    "package": 300.0,
    "deploy_demo": 180.0,
    "healthcheck": 120.0,
    "route": 60.0,
}

#: Healthcheck: 20 tries, 3 s apart, any 2xx/3xx is healthy.
HEALTHCHECK_RETRIES = 20
HEALTHCHECK_DELAY_SECONDS = 3.0

#: Live log batching: at most one event per stream per 500 ms, ≤ 200 lines.
LOG_FLUSH_SECONDS = 0.5
LOG_MAX_LINES = 200

#: The ledger entry that always means "the port the demo host points at".
DEMO_SERVICE = "demo"
#: Prefix for the transient port a deploy stages the replacement container on.
STAGING_SERVICE_PREFIX = "demo-staging-"

#: Bytes of a step's output kept in the database.
MAX_STORED_TAIL_CHARS = 65_536


@runtime_checkable
class ContainerRuntime(Protocol):
    """The container operations a pipeline needs.

    A structural type rather than a new port: ``DockerSandboxAdapter`` already
    implements it and tests inject a fake. ``SandboxProviderPort`` is frozen by
    the plan, so the extra capability lives here rather than in that contract.
    """

    async def run_step(
        self,
        *,
        organization_id: str,
        project_id: str,
        run_id: str,
        image: str,
        work_dir: str,
        command: str,
        timeout_seconds: float,
        on_output: Callable[[str], Awaitable[None]] | None = ...,
        role: str = ...,
        publish_port: int | None = ...,
        app_port: int | None = ...,
        keep_alive: bool = ...,
        name_suffix: str = ...,
    ) -> ProcessResult: ...

    async def remove_container(self, container_id: str) -> bool: ...


async def http_healthcheck(
    url: str,
    *,
    retries: int = HEALTHCHECK_RETRIES,
    delay_seconds: float = HEALTHCHECK_DELAY_SECONDS,
) -> bool:
    """Poll ``url`` until it answers 2xx/3xx, or give up.

    Deliberately tolerant of a slow start: a Next.js server can take 10-20 s to
    bind its port, and a healthcheck that gives up early turns a working deploy
    into a failed one.
    """
    import httpx

    remaining = retries
    async with httpx.AsyncClient(timeout=5.0, follow_redirects=False) as client:
        while remaining > 0:
            remaining -= 1
            try:
                response = await client.get(url)
            except httpx.HTTPError:
                response = None
            if response is not None and 200 <= response.status_code < 400:
                return True
            if remaining:
                await asyncio.sleep(delay_seconds)
    return False


async def _emit(*, type: str, organization_id: str, data: dict) -> None:
    """Publish one delivery event on the tenant channel. Never raises."""
    with contextlib.suppress(Exception):
        await get_publisher().publish(
            RealtimeEvent(
                type=type,
                channels=(org_channel(organization_id),),
                data=data,
                organization_id=organization_id,
            )
        )


class _LogBatcher:
    """Coalesce a step's output into ≤ 1 event / 500 ms, ≤ 200 lines each.

    Without this an ``npm install`` emits thousands of lines and drowns the
    socket. The batcher is per-step, so two steps never merge their output. It
    also keeps a bounded tail of everything it has seen, because the stored log
    must not lose the last lines to a flush that already emptied the buffer.
    """

    __slots__ = ("organization_id", "run_id", "step", "_buffer", "_tail", "_last_flush", "_lock")

    def __init__(self, *, organization_id: str, run_id: str, step: str) -> None:
        self.organization_id = organization_id
        self.run_id = run_id
        self.step = step
        self._buffer: list[str] = []
        self._tail = ""
        self._last_flush = time.monotonic()
        self._lock = asyncio.Lock()

    async def feed(self, chunk: str) -> None:
        if not chunk:
            return
        lines = chunk.splitlines() or [chunk]
        self._tail = (self._tail + "\n" + "\n".join(lines))[-MAX_STORED_TAIL_CHARS:]
        async with self._lock:
            self._buffer.extend(lines)
            now = time.monotonic()
            if now - self._last_flush >= LOG_FLUSH_SECONDS or len(self._buffer) >= LOG_MAX_LINES:
                await self._flush_locked()
                self._last_flush = now

    async def flush(self) -> None:
        async with self._lock:
            await self._flush_locked()

    @property
    def tail(self) -> str:
        """The stored log tail, redacted.

        Redaction happens on the whole accumulated tail rather than per chunk: a
        secret split across two reads of the pipe would slip past a per-chunk
        filter, and a build log is exactly where a credential ends up.
        """
        return redact_text(self._tail)

    async def _flush_locked(self) -> None:
        if not self._buffer:
            return
        lines, self._buffer = self._buffer, []
        # Redacted before publishing: a build log is exactly where a token pasted
        # into a URL, or an `env` dump, shows up.
        payload = redact_text("\n".join(lines))
        await _emit(
            type="pipeline.log",
            organization_id=self.organization_id,
            data={
                "run_id": self.run_id,
                "step": self.step,
                "lines": payload.splitlines()[:LOG_MAX_LINES],
            },
        )


@dataclass(slots=True)
class _StepOutcome:
    ok: bool
    error: str = ""


@dataclass(slots=True)
class _DeployContext:
    """State carried between steps within one run."""

    project_id: str = ""
    staging_port: int | None = None
    staging_service: str = ""
    new_container_id: str | None = None
    previous_container_id: str | None = None


@dataclass(slots=True)
class _CommandStep:
    """A step that simply runs a shell command in a fresh container."""

    name: str
    command: str
    timeout: float


class DeliveryService:
    """The delivery hooks, the pipeline executor and the demo runtime."""

    def __init__(
        self,
        *,
        runs: PipelineRunRepository,
        steps: PipelineStepRepository,
        releases: ReleaseRepository,
        runtimes: AppRuntimeRepository,
        projects: ProjectRepository,
        profiles: AppProfileRepository,
        organizations: OrganizationRepository,
        ports: PreviewDomainRepository,
        workspace: RepoWorkspacePort,
        domains: DomainRouterPort,
        containers: ContainerRuntime,
        artifacts_root: str | None = None,
        healthcheck: Callable[[str], Awaitable[bool]] | None = None,
    ) -> None:
        self.runs = runs
        self.steps = steps
        self.releases = releases
        self.runtimes = runtimes
        self.projects = projects
        self.profiles = profiles
        self.organizations = organizations
        self.ports = ports
        self.workspace = workspace
        self.domains = domains
        self.containers = containers
        self.artifacts_root = os.path.abspath(artifacts_root or settings.artifacts_root)
        self.healthcheck = healthcheck or http_healthcheck

    # ------------------------------------------------------------------ hooks

    async def on_app_created(self, *, organization_id: str, project_id: str) -> None:
        """Give a brand-new application its demo identity and its first build.

        Called by the org wizard after its transaction commits. Everything here
        is idempotent — a retry must not allocate a second port or a second host.
        """
        profile = await self.profiles.get(organization_id=organization_id, project_id=project_id)
        if profile is None:
            logger.warning("on_app_created: no app profile for project %s", project_id)
            return

        organization = await self.organizations.get(organization_id)
        if organization is None:
            return

        runtime = await self.runtimes.get(project_id)
        if runtime is None:
            port = await self.allocate_port(
                organization_id=organization_id, project_id=project_id, service_name=DEMO_SERVICE
            )
            host = build_host(organization.slug, settings.preview_root_domain)
            try:
                await self.domains.claim_preview_domain(
                    organization_id=organization_id,
                    project_id=project_id,
                    requested_subdomain=organization.slug,
                    port=port,
                )
            except Exception as exc:  # noqa: BLE001 - a taken host must not block the app
                logger.warning("demo host %s could not be claimed: %s", host, exc)

            runtime = await self.runtimes.upsert(
                AppRuntime(
                    project_id=project_id,
                    organization_id=organization_id,
                    demo_host=host,
                    host_port=port,
                    status=RuntimeStatus.PROVISIONING.value,
                )
            )
            await _emit(
                type="release.live",
                organization_id=organization_id,
                data={
                    "project_id": project_id,
                    "demo_host": host,
                    "demo_url": runtime.demo_url,
                    "status": "provisioning",
                },
            )

        # The workspace must exist before a run can check anything out.
        try:
            await self.workspace.ensure_workspace(project_id=project_id)
        except Exception as exc:  # noqa: BLE001 - a git outage must not fail the wizard
            logger.warning("ensure_workspace failed for project %s: %s", project_id, exc)
            return

        sha = await self.workspace.head(project_id=project_id, branch="dev")
        if sha:
            await self.queue_run(
                organization_id=organization_id,
                project_id=project_id,
                sha=sha,
                trigger=RunTrigger.ENGINE.value,
            )

    async def on_branch_pushed(
        self,
        *,
        organization_id: str,
        project_id: str,
        branch: str,
        sha: str,
        source: str,
    ) -> None:
        """Queue a build for a push, superseding any queued run for the project."""
        if branch != "dev":
            # Only `dev` is deployable; a task branch is work in progress.
            return
        trigger = RunTrigger.POLL.value if source == "poll" else RunTrigger.ENGINE.value
        await self.queue_run(
            organization_id=organization_id,
            project_id=project_id,
            sha=sha,
            trigger=trigger,
            branch=branch,
        )

    # ------------------------------------------------------------------ ports

    async def allocate_port(
        self, *, organization_id: str, project_id: str, service_name: str
    ) -> int:
        """Reserve a port from the demo range, database-backed.

        The router adapters keep their own in-memory ledger for preview domains,
        but the API and the delivery worker are separate processes — an
        in-process map would hand the same port to two projects as soon as a
        second process started. The `preview_ports` table is the shared truth.
        """
        existing = await self.ports.get_port(project_id, service_name)
        if existing is not None:
            return existing

        used = await self.ports.list_used_ports(
            start=settings.demo_port_start, end=settings.demo_port_end
        )
        for port in range(settings.demo_port_start, settings.demo_port_end + 1):
            if port in used:
                continue
            try:
                assert_port_allocatable(
                    port, start=settings.demo_port_start, end=settings.demo_port_end
                )
            except RuleViolation:
                continue
            await self.ports.add_port(
                organization_id=organization_id,
                project_id=project_id,
                service_name=service_name,
                port=port,
            )
            return port
        raise ConflictError("پورت آزادی در بازهٔ نمایشی موجود نیست.")

    # ------------------------------------------------------------------ queue

    async def queue_run(
        self,
        *,
        organization_id: str,
        project_id: str,
        sha: str,
        trigger: str = RunTrigger.ENGINE.value,
        branch: str = "dev",
    ) -> PipelineRun:
        """Create a queued run and demote older *queued* runs for the project.

        One active run per project: two builds racing for the same demo port is a
        mess with no upside. A run that has already started is left alone —
        cancelling it mid-flight would strand a container.
        """
        project = await self.projects.get(project_id)
        if project is None or project.organization_id != organization_id:
            raise NotFoundError("اپلیکیشن یافت نشد.")

        run = await self.runs.queue(
            PipelineRun(
                organization_id=organization_id,
                project_id=project_id,
                branch=branch,
                sha=sha,
                trigger=trigger,
                status=RunStatus.QUEUED.value,
            )
        )
        superseded = await self.runs.supersede_queued(project_id=project_id, keep_id=run.id)
        await _emit(
            type="pipeline.queued",
            organization_id=organization_id,
            data={
                "run_id": run.id,
                "project_id": project_id,
                "sha": sha,
                "trigger": trigger,
                "superseded": superseded,
            },
        )
        return run

    async def cancel_run(self, *, organization_id: str, run_id: str) -> PipelineRun:
        run = await self.runs.get_for_org(organization_id, run_id)
        if run is None:
            raise NotFoundError("اجرای خط لوله یافت نشد.")
        if run.is_terminal:
            raise ConflictError("این اجرا به پایان رسیده است.")
        await self.runs.set_status(
            run_id, RunStatus.CANCELLED.value, error_summary="لغو شد.", finished=True
        )
        refreshed = await self.runs.get(run_id) or run
        await self._emit_status(refreshed, error="لغو شد.")
        return refreshed

    async def request_manual_run(self, *, organization_id: str, project_id: str) -> PipelineRun:
        """Re-run the current `dev` head. Never a production deploy."""
        sha = await self.workspace.head(project_id=project_id, branch="dev")
        if not sha:
            raise ValidationError("شاخهٔ dev هنوز هیچ نسخه‌ای ندارد.")
        return await self.queue_run(
            organization_id=organization_id,
            project_id=project_id,
            sha=sha,
            trigger=RunTrigger.MANUAL.value,
        )

    # ------------------------------------------------------------------ execute

    async def execute_run(self, run_id: str) -> PipelineRun:
        """Run one queued pipeline end to end.

        Always returns a terminal run; a failure is recorded, never raised. The
        worker calls this in a loop and must not die because one build broke.
        """
        run = await self.runs.get(run_id)
        if run is None:
            raise NotFoundError("اجرای خط لوله یافت نشد.")
        if run.status != RunStatus.QUEUED.value:
            # Someone else claimed it, or it was superseded. Not an error.
            return run

        # Claim it atomically. Two workers polling the same queue would otherwise
        # both start a build and both publish a container to the demo port.
        if not await self.runs.claim(run_id):
            return await self.runs.get(run_id) or run
        run = await self.runs.get(run_id) or run

        profile = await self.profiles.get(
            organization_id=run.organization_id, project_id=run.project_id
        )
        runtime = await self.runtimes.get(run.project_id)
        if profile is None or runtime is None:
            return await self._fail_early(run, "پروفایل اپلیکیشن یا محیط نمایشی پیدا نشد.")

        steps = await self.steps.create_many(
            [
                PipelineStep(run_id=run_id, index=index, name=name)
                for index, name in enumerate(PIPELINE_STEPS)
            ]
        )
        await _emit(
            type="pipeline.status",
            organization_id=run.organization_id,
            data={
                "run_id": run_id,
                "project_id": run.project_id,
                "status": RunStatus.RUNNING.value,
                "steps": [s.name for s in steps],
            },
        )

        context = _DeployContext(
            project_id=run.project_id,
            previous_container_id=runtime.demo_container_id,
        )
        worktree = None
        try:
            worktree = await self.workspace.create_task_worktree(
                project_id=run.project_id, task_id=f"run-{run_id}", base=run.sha or "dev"
            )
            image = image_for_language(profile.language) or FALLBACK_IMAGE

            for step in steps:
                outcome = await self._run_step(
                    run=run,
                    step=step,
                    profile=profile,
                    runtime=runtime,
                    image=image,
                    work_dir=worktree.host_path,
                    context=context,
                )
                if not outcome.ok:
                    await self._cleanup_staging(context)
                    await self._fail_run(run, step, outcome.error)
                    return await self.runs.get(run_id) or run

            release = await self._publish_release(run, context)
            await self.runs.set_status(run_id, RunStatus.SUCCEEDED.value, finished=True)
            await self._emit_status(await self.runs.get(run_id) or run)
            refreshed_runtime = await self.runtimes.get(run.project_id) or runtime
            await _emit(
                type="release.live",
                organization_id=run.organization_id,
                data={
                    "run_id": run_id,
                    "project_id": run.project_id,
                    "release_id": release.id,
                    "sha": release.sha,
                    "demo_host": refreshed_runtime.demo_host,
                    "demo_url": refreshed_runtime.demo_url,
                },
            )
            return await self.runs.get(run_id) or run

        except asyncio.CancelledError:
            await self._cleanup_staging(context)
            await self.runs.set_status(
                run_id, RunStatus.CANCELLED.value, error_summary="لغو شد.", finished=True
            )
            raise
        except Exception as exc:  # noqa: BLE001 - a pipeline must never crash the worker
            logger.exception("pipeline run %s failed", run_id)
            await self._cleanup_staging(context)
            await self._fail_run(run, None, _plain_reason(exc))
            return await self.runs.get(run_id) or run
        finally:
            if worktree is not None:
                # A leftover worktree would collide with the next run's.
                with contextlib.suppress(Exception):
                    await self.workspace.remove_worktree(worktree=worktree)

    async def _fail_early(self, run: PipelineRun, reason: str) -> PipelineRun:
        await self.runs.set_status(
            run.id, RunStatus.FAILED.value, error_summary=reason, finished=True
        )
        refreshed = await self.runs.get(run.id) or run
        await self._emit_status(refreshed, error=reason)
        return refreshed

    # -- one step ------------------------------------------------------------

    async def _run_step(
        self,
        *,
        run: PipelineRun,
        step: PipelineStep,
        profile: AppProfile,
        runtime: AppRuntime,
        image: str,
        work_dir: str,
        context: _DeployContext,
    ) -> _StepOutcome:
        await self.steps.mark(step.id, StepStatus.RUNNING.value, started=True)
        await _emit(
            type="pipeline.step",
            organization_id=run.organization_id,
            data={
                "run_id": run.id,
                "step": step.name,
                "index": step.index,
                "status": StepStatus.RUNNING.value,
            },
        )
        batcher = _LogBatcher(
            organization_id=run.organization_id, run_id=run.id, step=step.name
        )

        command = _command_for(step.name, profile)
        if step.name in OPTIONAL_STEPS and not (command or "").strip():
            return await self._skip(
            step, batcher, "دستوری تنظیم نشده است", run.organization_id
        )

        if step.name == "healthcheck":
            return await self._healthcheck_step(run, step, profile, context, batcher)
        if step.name == "route":
            return await self._route_step(run, step, runtime, context, batcher)
        if step.name == "deploy_demo":
            return await self._deploy_demo_step(
                run, step, profile, runtime, image, work_dir, context, batcher
            )
        if step.name == "package":
            return await self._package_step(run, step, work_dir, batcher)

        result = await self.containers.run_step(
            organization_id=run.organization_id,
            project_id=run.project_id,
            run_id=f"{run.id}-{step.name}",
            image=image,
            work_dir=work_dir,
            command=command or "true",
            timeout_seconds=STEP_TIMEOUTS.get(step.name, 600.0),
            on_output=batcher.feed,
            role=ROLE_BUILD,
            name_suffix=step.name,
        )
        if result.stdout:
            await batcher.feed(result.stdout)
        if result.stderr:
            await batcher.feed(result.stderr)
        await batcher.flush()

        return await self._finish(
            step,
            batcher,
            organization_id=run.organization_id,
            ok=result.ok,
            exit_code=result.exit_code,
            error=f"مرحلهٔ «{step.name}» ناموفق بود"
            + (f" (کد {result.exit_code})." if result.exit_code is not None else "."),
        )

    async def _deploy_demo_step(
        self,
        run: PipelineRun,
        step: PipelineStep,
        profile: AppProfile,
        runtime: AppRuntime,
        image: str,
        work_dir: str,
        context: _DeployContext,
        batcher: _LogBatcher,
    ) -> _StepOutcome:
        """Start the replacement container on a *staging* port.

        The live port is deliberately not reused: the previous container still
        holds it, and reusing it would mean stopping the working demo before the
        new one is known to be healthy. That is the whole of blue/green.
        """
        staging_service = f"{STAGING_SERVICE_PREFIX}{run.id[:10]}"
        try:
            staging_port = await self.allocate_port(
                organization_id=run.organization_id,
                project_id=run.project_id,
                service_name=staging_service,
            )
        except ConflictError as exc:
            return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id, ok=False, error=str(exc))

        context.staging_port = staging_port
        context.staging_service = staging_service
        await batcher.feed(f"[deploy_demo] راه‌اندازی نسخهٔ جدید روی پورت {staging_port}\n")

        result = await self.containers.run_step(
            organization_id=run.organization_id,
            project_id=run.project_id,
            run_id=f"{run.id}-deploy",
            image=image,
            work_dir=work_dir,
            command=profile.start_command or "sleep infinity",
            timeout_seconds=STEP_TIMEOUTS["deploy_demo"],
            on_output=batcher.feed,
            role=ROLE_DEMO,
            publish_port=staging_port,
            app_port=profile.app_port or staging_port,
            keep_alive=True,
            name_suffix="demo",
        )
        if result.stdout:
            await batcher.feed(result.stdout)
        if result.stderr:
            await batcher.feed(result.stderr)
        await batcher.flush()

        if result.ok:
            context.new_container_id = (
                (result.stdout or "").strip().splitlines()[-1].strip() or None
            )

        return await self._finish(
            step,
            batcher,
            organization_id=run.organization_id,
            ok=result.ok,
            exit_code=result.exit_code,
            error="راه‌اندازی نسخهٔ نمایشی ناموفق بود.",
        )

    async def _healthcheck_step(
        self,
        run: PipelineRun,
        step: PipelineStep,
        profile: AppProfile,
        context: _DeployContext,
        batcher: _LogBatcher,
    ) -> _StepOutcome:
        """Poll the *new* container. No route switch happens unless this passes."""
        if not context.staging_port:
            return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id, ok=False, error="پورت آزمایشی تخصیص نشده است."
            )
        path = profile.healthcheck_path or "/"
        url = f"http://127.0.0.1:{context.staging_port}{path}"
        await batcher.feed(f"[healthcheck] بررسی {url}\n")
        healthy = await self.healthcheck(url)
        await batcher.feed(
            "[healthcheck] سرویس پاسخ داد.\n" if healthy else "[healthcheck] سرویس پاسخ نداد.\n"
        )
        await batcher.flush()
        return await self._finish(
            step,
            batcher,
            organization_id=run.organization_id,
            ok=healthy,
            exit_code=0 if healthy else 1,
            error="سرویس نمایشی به بررسی سلامت پاسخ نداد.",
        )

    async def _route_step(
        self,
        run: PipelineRun,
        step: PipelineStep,
        runtime: AppRuntime,
        context: _DeployContext,
        batcher: _LogBatcher,
    ) -> _StepOutcome:
        """Point the demo host at the new port, then retire the old container.

        This is the commit point of the whole run. Everything before it can fail
        without consequence; nothing after it is allowed to.
        """
        if not runtime.demo_host or not context.staging_port:
            return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id,
                ok=False,
                error="میزبان یا پورت نمایشی تنظیم نشده است.",
            )

        result = await self.domains.configure_reverse_proxy(
            host=runtime.demo_host, port=context.staging_port
        )
        await batcher.feed(f"[route] {result.message}\n")
        if not result.ok:
            await batcher.flush()
            return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id, ok=False, error="تنظیم مسیر دامنه ناموفق بود."
            )

        # The switch. Order matters: `preview_ports.port` is unique platform-wide,
        # so the transient staging reservation has to be released *before* the
        # `demo` entry can be re-pointed at the port it was holding.
        await self.ports.delete_port(
            project_id=run.project_id, service_name=context.staging_service
        )
        await self.ports.set_port(
            project_id=run.project_id, service_name=DEMO_SERVICE, port=context.staging_port
        )
        await self.runtimes.set_live(
            run.project_id,
            container_id=context.new_container_id,
            host_port=context.staging_port,
            release_id=None,
            status=RuntimeStatus.LIVE.value,
        )

        # Only now is it safe to stop the previous container.
        if (
            context.previous_container_id
            and context.previous_container_id != context.new_container_id
        ):
            await batcher.feed(
                f"[route] توقف نسخهٔ قبلی ({context.previous_container_id[:12]})\n"
            )
            with contextlib.suppress(Exception):
                await self.containers.remove_container(context.previous_container_id)
            context.previous_container_id = None

        await batcher.flush()
        return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id, ok=True, exit_code=0)

    async def _package_step(
        self, run: PipelineRun, step: PipelineStep, work_dir: str, batcher: _LogBatcher
    ) -> _StepOutcome:
        """Tar the run directory into ``ARTIFACTS_ROOT/<org>/<project>/<sha7>.tar.gz``.

        Excludes ``.git`` and ``node_modules``: the artifact is a deployable
        build, not a copy of the repository. A packaging failure is recorded but
        does not fail the run — the demo serves from the work tree, and a missing
        artifact only costs a production deploy its source.
        """
        target_dir = os.path.join(self.artifacts_root, run.organization_id, run.project_id)
        os.makedirs(target_dir, exist_ok=True)
        sha7 = (run.sha or "unknown")[:7]
        artifact = os.path.join(target_dir, f"{sha7}.tar.gz")

        result = await self.containers.run_step(
            organization_id=run.organization_id,
            project_id=run.project_id,
            run_id=f"{run.id}-package",
            image=image_for_language(None),
            work_dir=work_dir,
            command=(
                "tar --exclude=./.git --exclude=./node_modules --exclude=./.next/cache "
                f"-czf {artifact!r} -C {work_dir!r} ."
            ),
            timeout_seconds=STEP_TIMEOUTS["package"],
            on_output=batcher.feed,
            role=ROLE_BUILD,
            name_suffix="package",
        )
        if result.stdout:
            await batcher.feed(result.stdout)
        if result.stderr:
            await batcher.feed(result.stderr)
        await batcher.flush()
        if not result.ok:
            await batcher.feed("[package] بسته‌بندی ناموفق بود؛ اجرا ادامه می‌یابد.\n")
            await batcher.flush()
        return await self._finish(
                step,
                batcher,
                organization_id=run.organization_id, ok=True, exit_code=result.exit_code)

    async def _skip(
        self, step: PipelineStep, batcher: _LogBatcher, reason: str, organization_id: str
    ) -> _StepOutcome:
        await batcher.feed(f"[{step.name}] {reason}؛ رد شد.\n")
        await batcher.flush()
        await self.steps.mark(
            step.id, StepStatus.SKIPPED.value, log_tail=batcher.tail, finished=True
        )
        await self._emit_step(step, StepStatus.SKIPPED.value, organization_id)
        return _StepOutcome(ok=True)

    async def _finish(
        self,
        step: PipelineStep,
        batcher: _LogBatcher,
        *,
        ok: bool,
        organization_id: str,
        exit_code: int | None = None,
        error: str = "",
    ) -> _StepOutcome:
        status = StepStatus.SUCCEEDED.value if ok else StepStatus.FAILED.value
        await self.steps.mark(
            step.id, status, exit_code=exit_code, log_tail=batcher.tail, finished=True
        )
        await self._emit_step(step, status, organization_id)
        return _StepOutcome(ok=ok, error=error if not ok else "")

    # -- release -------------------------------------------------------------

    async def _publish_release(self, run: PipelineRun, context: _DeployContext) -> Release:
        sha7 = (run.sha or "unknown")[:7]
        artifact = os.path.join(
            self.artifacts_root, run.organization_id, run.project_id, f"{sha7}.tar.gz"
        )
        release = await self.releases.create(
            Release(
                organization_id=run.organization_id,
                project_id=run.project_id,
                run_id=run.id,
                sha=run.sha,
                artifact_path=artifact,
            )
        )
        await self.releases.mark_live(project_id=run.project_id, release_id=release.id)
        # The route step already made the runtime live; bind it to the release now
        # that one exists.
        await self.runtimes.set_live(
            run.project_id,
            container_id=context.new_container_id,
            host_port=context.staging_port,
            release_id=release.id,
            status=RuntimeStatus.LIVE.value,
        )
        return release

    async def _cleanup_staging(self, context: _DeployContext) -> None:
        """Remove a failed replacement and release its port.

        The *previous* container is never touched: a failed run must leave the
        demo exactly as it found it.
        """
        if context.new_container_id:
            with contextlib.suppress(Exception):
                await self.containers.remove_container(context.new_container_id)
            context.new_container_id = None
        if context.staging_service and context.project_id:
            with contextlib.suppress(Exception):
                await self.ports.delete_port(
                    project_id=context.project_id, service_name=context.staging_service
                )

    async def _fail_run(
        self, run: PipelineRun, step: PipelineStep | None, error: str
    ) -> None:
        await self.runs.set_status(
            run.id, RunStatus.FAILED.value, error_summary=error, finished=True
        )
        await self.runtimes.set_status(run.project_id, RuntimeStatus.DEGRADED.value)
        refreshed = await self.runs.get(run.id) or run
        await self._emit_status(refreshed, error=error)
        logger.info(
            "run %s failed at %s: %s", run.id, step.name if step else "setup", error
        )

    async def _emit_step(self, step: PipelineStep, status: str, organization_id: str) -> None:
        await _emit(
            type="pipeline.step",
            organization_id=organization_id,
            data={
                "run_id": step.run_id,
                "step": step.name,
                "index": step.index,
                "status": status,
            },
        )

    async def _emit_status(self, run: PipelineRun, *, error: str = "") -> None:
        await _emit(
            type="pipeline.status",
            organization_id=run.organization_id,
            data={
                "run_id": run.id,
                "project_id": run.project_id,
                "status": run.status,
                "error": error,
            },
        )

    # ------------------------------------------------------------------ reads

    async def list_runs(
        self, *, organization_id: str, project_id: str, limit: int = 50, offset: int = 0
    ) -> list[PipelineRun]:
        return await self.runs.list_for_project(
            organization_id=organization_id, project_id=project_id, limit=limit, offset=offset
        )

    async def get_run_detail(
        self, *, organization_id: str, run_id: str
    ) -> tuple[PipelineRun, list[PipelineStep]]:
        run = await self.runs.get_for_org(organization_id, run_id)
        if run is None:
            raise NotFoundError("اجرای خط لوله یافت نشد.")
        return run, await self.steps.list_for_run(run_id)

    async def list_releases(self, *, organization_id: str, project_id: str) -> list[Release]:
        project = await self.projects.get(project_id)
        if project is None or project.organization_id != organization_id:
            raise NotFoundError("اپلیکیشن یافت نشد.")
        return await self.releases.list_for_project(project_id=project_id)

    async def runtime_status(self, *, organization_id: str, project_id: str) -> AppRuntime:
        runtime = await self.runtimes.get_for_org(organization_id, project_id)
        if runtime is None:
            raise NotFoundError("محیط نمایشی یافت نشد.")
        return runtime

    async def rollback_demo(
        self, *, organization_id: str, project_id: str, release_id: str | None = None
    ) -> Release:
        """Point the demo back at an earlier release.

        The route moves; the container does not. A rollback that rebuilt would
        not be a rollback — it would be a deploy of code that is already known to
        be older than what is live.
        """
        releases = await self.releases.list_for_project(project_id=project_id)
        if not releases:
            raise NotFoundError("نسخه‌ای برای بازگردانی وجود ندارد.")

        target: Release | None = None
        if release_id is not None:
            target = next((r for r in releases if r.id == release_id), None)
        else:
            live = next((r for r in releases if r.is_demo_live), None)
            target = next((r for r in releases if live is None or r.id != live.id), None)

        if target is None or target.organization_id != organization_id:
            raise NotFoundError("نسخه یافت نشد.")

        runtime = await self.runtimes.get(project_id)
        if runtime is None:
            raise NotFoundError("محیط نمایشی یافت نشد.")
        if runtime.demo_host and runtime.host_port:
            await self.domains.configure_reverse_proxy(
                host=runtime.demo_host, port=runtime.host_port
            )
        await self.releases.mark_live(project_id=project_id, release_id=target.id)
        await self.runtimes.set_live(
            project_id,
            container_id=runtime.demo_container_id,
            host_port=runtime.host_port,
            release_id=target.id,
            status=RuntimeStatus.LIVE.value,
        )
        await _emit(
            type="release.live",
            organization_id=organization_id,
            data={
                "project_id": project_id,
                "release_id": target.id,
                "sha": target.sha,
                "rolled_back": True,
                "demo_host": runtime.demo_host,
            },
        )
        return target

    # ------------------------------------------------------------------ polling

    async def poll_remote(self, *, organization_id: str, project_id: str) -> PipelineRun | None:
        """Build a push that happened outside the engine.

        A user pushing to their own git is invisible to us otherwise: the engine
        only tells us about pushes it makes itself.
        """
        head = await self.workspace.head(project_id=project_id, branch="dev")
        if not head:
            return None
        last = await self.runs.last_succeeded_sha(project_id)
        if last == head:
            return None
        active = await self.runs.active_for_project(project_id)
        if active is not None and active.sha == head:
            return None
        return await self.queue_run(
            organization_id=organization_id,
            project_id=project_id,
            sha=head,
            trigger=RunTrigger.POLL.value,
        )


def _command_for(step: str, profile: AppProfile) -> str:
    return {
        "checkout": "git rev-parse HEAD && git status --porcelain=v1 -uall | head -50",
        "setup": profile.setup_command or "",
        "build": profile.build_command or "",
        "test": profile.test_command or "",
    }.get(step, "")


def _plain_reason(exc: BaseException) -> str:
    """A redacted, human-readable one-liner. Never a stack trace, never a secret."""
    return redact_text(str(exc) or type(exc).__name__)[:400]


__all__ = [
    "DEMO_SERVICE",
    "HEALTHCHECK_DELAY_SECONDS",
    "HEALTHCHECK_RETRIES",
    "LOG_FLUSH_SECONDS",
    "LOG_MAX_LINES",
    "MAX_STORED_TAIL_CHARS",
    "STAGING_SERVICE_PREFIX",
    "STEP_TIMEOUTS",
    "ContainerRuntime",
    "DeliveryService",
    "http_healthcheck",
]