"""PipelineWatcher: after a push — wait for the uploads CI, then deploy through tgit-runner (SPEC C-24, C-30…C-34).

The truth is in the DB (`pushes.ci_status/cd_status`, `pipeline_runs`), so a restarted bot picks up where it stopped.
The CI is polled only while some push waits for it (ADR-003); with nothing to wait for the loop sleeps until `wake()`.

The push message is edited in place (CI ⏳ → ✅/❌ → CD ⏳ → 🚀/⛔); the author also gets separate messages for the
outcomes — an edit doesn't notify. A failed deploy is an alert ⛔ for the whole team (raised by the Deployer).
"""

import asyncio
import contextlib
import html
import logging
from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta

from aiogram import Bot
from aiogram.enums import ParseMode
from aiogram.exceptions import TelegramAPIError, TelegramBadRequest
from aiogram.types import BufferedInputFile, ReplyParameters
from sqlalchemy import or_, select, update

from app.db.models import PipelineRun, PipelineStatus, Push, Stage, as_utc, utcnow
from app.db.session import SessionMaker
from app.pipeline.provider import CIProvider, CIRun
from app.runner.deployer import Deployer, Outcome, changed_services
from app.texts import Texts

log = logging.getLogger(__name__)

WAITING = (PipelineStatus.PENDING, PipelineStatus.RUNNING)
PASSED = (PipelineStatus.SUCCESS, PipelineStatus.SKIPPED)
RUNNER = "tgit-runner"  # run_slug of the CD stage: the deploy is the bot's own, not a CI run
LOG_CHARS = 3000  # inline tail; the whole log goes as a file when it is longer
MAX_LOG_BYTES = 5 * 1024 * 1024  # SPEC L-5


def initial_status(ci: bool, deploys: bool, paths: list[str]) -> tuple[PipelineStatus, PipelineStatus]:
    """(ci, cd) of a fresh push: SKIPPED where there is nothing to wait for."""
    cd = PipelineStatus.PENDING if deploys and changed_services(paths) else PipelineStatus.SKIPPED
    return (PipelineStatus.PENDING if ci else PipelineStatus.SKIPPED), cd


def render(
    texts: Texts,
    push: Push,
    ci_run: PipelineRun | None = None,
    cd_run: PipelineRun | None = None,
    commit_url_template: str = "",
    timeout_minutes: float = 30,
) -> str:
    """The push message: what was pushed, then a line per stage that applies to it."""
    t = texts.pipeline
    first_line = push.message.splitlines()[0] if push.message else ""
    lines = [texts.commit.pushed.format(sha=push.sha[:7], message=first_line)]
    if commit_url_template:
        lines.append(commit_url_template.format(sha=push.sha))
    match push.ci_status:
        case PipelineStatus.PENDING:
            lines.append(t.ci_pending)
        case PipelineStatus.RUNNING:
            lines.append(t.ci_running)
        case PipelineStatus.SUCCESS:
            lines.append(t.ci_success)
        case PipelineStatus.FAILED:
            lines.append(t.ci_failed.format(step=(ci_run and ci_run.failed_cube) or "?"))
        case PipelineStatus.TIMEOUT:
            lines.append(t.ci_timeout.format(minutes=round(timeout_minutes)))
    if ci_run is not None and ci_run.url:
        lines.append(ci_run.url)

    services = ", ".join(changed_services(push.paths))
    failed = (cd_run.failed_cube or "") if cd_run else ""
    match push.cd_status:
        case PipelineStatus.PENDING if push.ci_status in WAITING:
            lines.append(t.cd_waiting.format(services=services))
        case PipelineStatus.PENDING | PipelineStatus.RUNNING:
            lines.append(t.cd_running.format(services=services))
        case PipelineStatus.SUCCESS:
            lines.append(t.cd_success.format(services=services))
        case PipelineStatus.FAILED:
            lines.append(t.cd_failed.format(services=failed.replace(",", ", ")))
        case PipelineStatus.SKIPPED if services and push.ci_status not in PASSED:
            lines.append(t.cd_not_run)
        case PipelineStatus.SKIPPED if cd_run is not None:
            lines.append(t.cd_superseded)
    return "\n".join(lines)


class PipelineWatcher:
    def __init__(
        self,
        sessions: SessionMaker,
        bot: Bot,
        texts: Texts,
        ci: CIProvider | None,
        deployer: Deployer | None,
        *,
        commit_url_template: str = "",
        poll_interval: float = 10.0,
        max_poll_interval: float = 60.0,
        run_timeout: timedelta = timedelta(minutes=30),
        clock: Callable[[], datetime] = utcnow,
    ) -> None:
        self._sessions = sessions
        self._bot = bot
        self._texts = texts
        self._ci = ci
        self._deployer = deployer
        self._url = commit_url_template
        self._interval = poll_interval
        self._max_interval = max_poll_interval
        self._timeout = run_timeout
        self._clock = clock
        self._wake = asyncio.Event()
        self._deploying: dict[int, asyncio.Task[None]] = {}

    def initial_status(self, paths: list[str]) -> tuple[PipelineStatus, PipelineStatus]:
        return initial_status(self._ci is not None, self._deployer is not None, paths)

    def wake(self) -> None:
        """A new push: look at it now rather than after the current pause."""
        self._wake.set()

    async def run(self) -> None:
        delay = self._interval
        while True:
            self._wake.clear()  # before reading the DB: a push recorded during the tick wakes the next wait
            try:
                polling = await self.tick()
                delay = self._interval
            except Exception:
                log.exception("pipeline watcher tick")
                polling = True
                delay = min(delay * 2, self._max_interval)
            if polling:
                with contextlib.suppress(TimeoutError):
                    await asyncio.wait_for(self._wake.wait(), timeout=delay)
            else:
                await self._wake.wait()

    async def tick(self) -> bool:
        """One pass over unfinished pushes. True — some push still waits for CI, poll again soon."""
        async with self._sessions() as s:
            pushes = list(
                await s.scalars(
                    select(Push)
                    .where(
                        or_(
                            Push.ci_status.in_(WAITING),
                            Push.ci_status.in_(PASSED) & Push.cd_status.in_(WAITING),
                        )
                    )
                    .order_by(Push.id)
                )
            )
        polling = False
        for push in pushes:
            if push.ci_status in WAITING:
                await self._poll_ci(push)
                polling |= push.ci_status in WAITING
            elif push.id not in self._deploying:
                await self._start_cd(push)
        return polling

    async def drain(self) -> None:
        """Wait for the deploys in flight."""
        while self._deploying:
            await asyncio.gather(*self._deploying.values(), return_exceptions=True)

    async def aclose(self) -> None:
        """Stop deploys in flight: their push stays `cd=running` in the DB, and the next start deploys it again."""
        tasks = list(self._deploying.values())
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

    # ---- CI ------------------------------------------------------------------------------------------------------

    async def _poll_ci(self, push: Push) -> None:
        if self._ci is None:  # CI switched off since the push: don't keep it waiting forever
            await self._set(push, ci=PipelineStatus.SKIPPED)
            await self._refresh(push)
            return
        run = await self._ci.find(push.sha)
        if run is None or run.status in WAITING:
            if self._clock() - as_utc(push.created_at) > self._timeout:
                await self._finish_ci(push, PipelineStatus.TIMEOUT, run)
            elif run is not None and run.status is not push.ci_status:
                await self._save_run(push, Stage.CI, run.id, run.status, None, run.url)
                await self._set(push, ci=run.status)
                await self._refresh(push)
            return
        await self._finish_ci(push, run.status, run)

    async def _finish_ci(self, push: Push, status: PipelineStatus, run: CIRun | None) -> None:
        if run is not None:
            await self._save_run(push, Stage.CI, run.id, status, run.failed_step, run.url)
        passed = status in PASSED
        cd = push.cd_status if passed or push.cd_status is not PipelineStatus.PENDING else PipelineStatus.SKIPPED
        await self._set(push, ci=status, cd=cd)
        await self._refresh(push)
        t = self._texts.pipeline
        sha = push.sha[:7]
        url = html.escape(run.url) if run else ""
        if passed:
            await self._notify(push, t.notify_ci_ok.format(sha=sha, url=url))
            if push.cd_status is PipelineStatus.PENDING:
                await self._start_cd(push)
        elif status is PipelineStatus.TIMEOUT:
            minutes = round(self._timeout.total_seconds() / 60)
            await self._notify(push, t.notify_ci_timeout.format(sha=sha, minutes=minutes))
        else:
            assert run is not None
            await self._notify_failure(push, run)

    async def _notify_failure(self, push: Push, run: CIRun) -> None:
        t = self._texts.pipeline
        step = html.escape(run.failed_step or "?")
        text = t.notify_ci_failed.format(step=step, sha=push.sha[:7], url=html.escape(run.url))
        assert self._ci is not None
        try:
            full = await self._ci.failure_log(run)
        except Exception as e:  # the verdict matters more than its log
            log.warning("ci log for %s: %s", push.sha[:7], e)
            full = ""
        tail = full.strip()[-LOG_CHARS:]
        if tail:
            text += f"\n<pre>{html.escape(tail)}</pre>"
        await self._notify(push, text)
        if len(full.strip()) > LOG_CHARS:
            data = full.encode()[-MAX_LOG_BYTES:]
            await self._send(
                lambda: self._bot.send_document(
                    push.chat_id,
                    BufferedInputFile(data, f"ci-{push.sha[:7]}.log"),
                    caption=t.log_file,
                    reply_parameters=self._reply(push),
                )
            )

    # ---- CD ------------------------------------------------------------------------------------------------------

    async def _start_cd(self, push: Push) -> None:
        if self._deployer is None:  # the runner was switched off since the push
            await self._set(push, cd=PipelineStatus.SKIPPED)
            await self._refresh(push)
            return
        services = await self._not_superseded(push)
        if not services:
            await self._save_run(push, Stage.CD, RUNNER, PipelineStatus.SKIPPED, None, None)
            await self._set(push, cd=PipelineStatus.SKIPPED)
            await self._refresh(push)
            return
        await self._set(push, cd=PipelineStatus.RUNNING)
        await self._refresh(push)
        task = asyncio.create_task(self._deploy(push, services))
        self._deploying[push.id] = task
        task.add_done_callback(lambda _: self._deploying.pop(push.id, None))

    async def _not_superseded(self, push: Push) -> list[str]:
        """Services of this push that no newer passed push will deploy anyway (CI runs may finish out of order)."""
        services = changed_services(push.paths)
        async with self._sessions() as s:
            newer = await s.scalars(select(Push.paths).where(Push.id > push.id, Push.ci_status.in_(PASSED)))
            taken = {svc for paths in newer for svc in changed_services(paths)}
        return [svc for svc in services if svc not in taken]

    async def _deploy(self, push: Push, services: list[str]) -> None:
        assert self._deployer is not None
        outcomes: list[Outcome] = []

        async def report(o: Outcome) -> None:
            outcomes.append(o)
            t = self._texts.pipeline
            name = html.escape(o.service)
            if o.result.ok and o.removed:
                await self._notify(push, t.removed.format(service=name))
            elif not o.result.ok:
                text = t.deploy_failed.format(service=name, reason=html.escape(o.result.summary))
                if o.result.tail:
                    text += f"\n<pre>{html.escape(o.result.tail[-LOG_CHARS:])}</pre>"
                await self._notify(push, text)

        try:
            await self._deployer.deploy(push.sha, services, report)
        except Exception:
            log.exception("deploy of push %s", push.sha[:7])
        failed = [o.service for o in outcomes if not o.result.ok] + services[len(outcomes) :]
        deployed = [o.service for o in outcomes if o.result.ok and not o.removed]
        status = PipelineStatus.FAILED if failed else PipelineStatus.SUCCESS
        await self._save_run(push, Stage.CD, RUNNER, status, ",".join(failed)[:100] or None, None)
        await self._set(push, cd=status)
        await self._refresh(push)
        if deployed:
            names = ", ".join(deployed)
            await self._notify(push, self._texts.pipeline.notify_cd_ok.format(services=names, sha=push.sha[:7]))

    # ---- state and messages --------------------------------------------------------------------------------------

    async def _set(self, push: Push, ci: PipelineStatus | None = None, cd: PipelineStatus | None = None) -> None:
        values = {}
        if ci is not None:
            values["ci_status"] = push.ci_status = ci
        if cd is not None:
            values["cd_status"] = push.cd_status = cd
        async with self._sessions.begin() as s:
            await s.execute(update(Push).where(Push.id == push.id).values(**values))

    async def _save_run(
        self, push: Push, stage: Stage, slug: str, status: PipelineStatus, failed: str | None, url: str | None
    ) -> None:
        async with self._sessions.begin() as s:
            row = await s.scalar(select(PipelineRun).where(PipelineRun.push_id == push.id, PipelineRun.stage == stage))
            if row is None:
                row = PipelineRun(push_id=push.id, stage=stage)
                s.add(row)
            row.run_slug, row.status, row.failed_cube, row.url = slug, status, failed, url

    async def _runs(self, push: Push) -> dict[Stage, PipelineRun]:
        async with self._sessions() as s:
            rows = await s.scalars(select(PipelineRun).where(PipelineRun.push_id == push.id))
            return {r.stage: r for r in rows}

    async def _refresh(self, push: Push) -> None:
        """Re-render the push message from the DB."""
        if push.message_id is None:
            return
        runs = await self._runs(push)
        text = render(
            self._texts,
            push,
            runs.get(Stage.CI),
            runs.get(Stage.CD),
            self._url,
            self._timeout.total_seconds() / 60,
        )
        message_id = push.message_id
        await self._send(lambda: self._bot.edit_message_text(text, chat_id=push.chat_id, message_id=message_id))

    def _reply(self, push: Push) -> ReplyParameters | None:
        if push.message_id is None:
            return None
        return ReplyParameters(message_id=push.message_id, allow_sending_without_reply=True)

    async def _notify(self, push: Push, text: str) -> None:
        await self._send(
            lambda: self._bot.send_message(
                push.chat_id, text, parse_mode=ParseMode.HTML, reply_parameters=self._reply(push)
            )
        )

    @staticmethod
    async def _send(call: Callable[[], Awaitable[object]]) -> None:
        """The status in the DB is what matters: a message that didn't get through is logged, not retried."""
        try:
            await call()
        except TelegramBadRequest as e:  # not modified · deleted by the user · too old to edit
            log.info("pipeline message: %s", e)
        except TelegramAPIError as e:  # blocked by the user · flood control
            log.warning("pipeline message: %s", e)