gitea-dynamic-runner/src/gitea_dynamic_runner/models.py

"""Validated messages shared by the controller and runner backends."""

from __future__ import annotations

from dataclasses import asdict, dataclass
import json


BACKENDS = frozenset({"pod", "vm"})
REQUIRED_LABEL = "self-hosted"


@dataclass(frozen=True, slots=True)
class RunnerRequest:
    """A queued Gitea job that needs one disposable runner environment."""

    job_id: int
    run_id: int
    backend: str
    repository: str
    job_name: str
    labels: tuple[str, ...]
    schema_version: int = 1

    @classmethod
    def from_webhook(cls, payload: object) -> RunnerRequest | None:
        if not isinstance(payload, dict) or payload.get("action") != "queued":
            return None

        job = payload.get("workflow_job")
        repository = payload.get("repository")
        if not isinstance(job, dict) or not isinstance(repository, dict):
            return None

        labels_value = job.get("labels")
        if not isinstance(labels_value, list) or not all(
            isinstance(label, str) for label in labels_value
        ):
            return None
        labels = tuple(dict.fromkeys(labels_value))
        selected = BACKENDS.intersection(labels)
        if REQUIRED_LABEL not in labels or len(selected) != 1:
            return None

        job_id = job.get("id")
        run_id = job.get("run_id")
        job_name = job.get("name")
        full_name = repository.get("full_name")
        if not _positive_int(job_id) or not _positive_int(run_id):
            return None
        if not all(_nonempty(value) for value in (job_name, full_name)):
            return None

        return cls(
            job_id=job_id,
            run_id=run_id,
            backend=next(iter(selected)),
            repository=full_name.strip(),
            job_name=job_name.strip(),
            labels=labels,
        )

    def to_json(self) -> bytes:
        document = asdict(self)
        document["labels"] = list(self.labels)
        return json.dumps(
            document,
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode()

    @classmethod
    def from_json(cls, body: bytes) -> RunnerRequest:
        document = json.loads(body)
        if not isinstance(document, dict) or document.get("schema_version") != 1:
            raise ValueError("unsupported runner request")
        labels = document.get("labels")
        if not isinstance(labels, list) or not all(
            isinstance(label, str) for label in labels
        ):
            raise ValueError("invalid runner labels")
        try:
            request = cls(
                job_id=document["job_id"],
                run_id=document["run_id"],
                backend=document["backend"],
                repository=document["repository"],
                job_name=document["job_name"],
                labels=tuple(labels),
            )
        except KeyError as error:
            raise ValueError(f"missing runner request field: {error.args[0]}") from error
        if (
            not _positive_int(request.job_id)
            or not _positive_int(request.run_id)
            or request.backend not in BACKENDS
            or not _nonempty(request.repository)
            or not _nonempty(request.job_name)
            or REQUIRED_LABEL not in request.labels
            or request.backend not in request.labels
        ):
            raise ValueError("invalid runner request")
        return request


@dataclass(frozen=True, slots=True)
class IdentityBinding:
    """The actual task claimed by an ephemeral runner."""

    job_id: int
    run_id: int
    backend: str
    runner_name: str
    repository: str
    job_name: str
    schema_version: int = 1

    @classmethod
    def from_webhook(cls, payload: object) -> IdentityBinding | None:
        if not isinstance(payload, dict) or payload.get("action") != "in_progress":
            return None
        job = payload.get("workflow_job")
        repository = payload.get("repository")
        if not isinstance(job, dict) or not isinstance(repository, dict):
            return None

        labels = job.get("labels")
        if not isinstance(labels, list) or not all(
            isinstance(label, str) for label in labels
        ):
            return None
        selected = BACKENDS.intersection(labels)
        if REQUIRED_LABEL not in labels or len(selected) != 1:
            return None
        backend = next(iter(selected))

        job_id = job.get("id")
        run_id = job.get("run_id")
        runner_name = job.get("runner_name")
        job_name = job.get("name")
        full_name = repository.get("full_name")
        if not _positive_int(job_id) or not _positive_int(run_id):
            return None
        if not all(
            _nonempty(value)
            for value in (runner_name, job_name, full_name)
        ):
            return None
        if not runner_name.startswith(f"gitea-{backend}-"):
            return None

        return cls(
            job_id=job_id,
            run_id=run_id,
            backend=backend,
            runner_name=runner_name.strip(),
            repository=full_name.strip(),
            job_name=job_name.strip(),
        )

    def to_json(self) -> bytes:
        return json.dumps(
            asdict(self),
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode()


def _positive_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool) and value > 0


def _nonempty(value: object) -> bool:
    return isinstance(value, str) and bool(value.strip())