platform-ci/platform_ci/effective_version.py

"""The effective version of a release build, calculated BEFORE the single native build from the declared version and the event's branch or tag.

UNTESTED: written while test execution was suspended. A rule runs ONLY where every input the captured calculator used is an immutable fact this contract carries; otherwise it FAILS
with the exact missing input. It never runs a different lifecycle.

The captured `calculate_version.sh` (run by `qrun-maven-prepare.mjs` before verify, for every eligible publish kind) rewrote the POM's `<revision>` inside the build job. Here the same
rules are a pure function: the repository's channel names one rule, `plan_targets` appends `-D<property>=<effective>` to the Maven commands, and so the effective version is part of the
target (its `config_digest`), is what the build consumes and records (`build_inputs.version`), and is recomputed by the trusted publisher and tag planner from forge facts (the declared
version at the release commit, the branch or tag). An artifact is never relabelled. Nothing here reads a repository or a file.

Rules (the captured handler of each branch type, selected by the repository's own channel, no name in code):

* `develop-next-minor`: branch `develop`. A declared RC or stable version becomes `X.(Y+1).0-SNAPSHOT`. A declared SNAPSHOT (plain or feature-format) follows the captured handler: it becomes
  `X.(Y+1).0-SNAPSHOT` when a commit reachable from the release commit, with a committer date at or after (build start - 3 days), has a message line matching the captured release-merge or
  "Bump version after release" predicate (`RELEASE_ACTIVITY_PATTERNS`), and is kept otherwise. The two inputs are immutable facts, never a clock read here: the build start is the SERVER's `started_at`
  of the run's one prepare job (the logical build start, as the captured `readJobStart` used the verify job's), and the history is the forge's commit listing of the release commit. The caller supplies
  them as `release_activity`, a callable that answers the question; without it a SNAPSHOT is still REFUSED (the outcome is unknown, never assumed to be `keep`).
* `rc-from-branch`: branch `release/...` (the captured unanchored `release/([0-9]+)[.]([0-9]+)` search). `X.Y.0-RC.n`, `n` one more than a declared `X.Y.Z-RC.m`, otherwise 1. With
  `rc_baseline: reconciled` (the legacy qbit profile, `reconciledRcBaseline`) a declared `X.Y.0-SNAPSHOT` is REFUSED: whether a tagged first-parent RC counter must be restored needs the commit's
  parents, their POMs and tag ancestry, none of which this contract carries.
* `patch-bump`: branch `hotfix/...`. Declared `X.Y.Z[...]` becomes `X.Y.(Z+1)`.
* `stable-from-tag`: pushed tag `vX.Y.Z` becomes `X.Y.Z` (the `stable` tag policy's release kind).
* `feature-snapshot`: ALWAYS REFUSED. The captured version embeds the originating feature branch of a `publish*` tag (`git branch -r --contains HEAD`) and git's abbreviated commit id; neither is a
  carried input. No fallback name or fixed-length hash is substituted.

Every rule needs a declared version in one of the captured formats (RC, SNAPSHOT, feature SNAPSHOT, stable), as the captured `extract_version_parts` did (it exited 1 otherwise).

Not carried at all (no captured repository fact enables them): `release_on_main` / main without a tag (latest tag by `git describe`) and the `v-prefix` RC-tag route. The calculator's `versions:set`
fallback when `project.version` does not use the property is refused by the plan instead.
"""

import re
from collections.abc import Callable, Iterable, Sequence
from typing import Final

from platform_ci.workspace import PipelineError


class ReleaseActivityError(PipelineError):
   """The release-activity input of a develop SNAPSHOT could not be established (no trusted build start, or history not proven); the version is unknown, never guessed."""


RELEASE_ACTIVITY_WINDOW_SECONDS: Final = 3 * 24 * 60 * 60
"""The captured `git log --since="3 days ago"` window, measured back from the build start (the captured prepare's `WINDOW_SECONDS`)."""
RELEASE_ACTIVITY_PATTERNS: Final = (
   re.compile(r"Merge.*release/[0-9]+[.][0-9]+([.][0-9]+)?([^A-Za-z0-9_./-]|$).*into.*develop"),
   re.compile(r"Merge.*release/[0-9]+[.][0-9]+([.][0-9]+)?([^A-Za-z0-9_./-]|$).*back.*develop"),
   re.compile(r"Bump.*version.*after.*release.*v"),
)
"""The captured `git log --extended-regexp --grep` predicates, exactly: the two numeric `release/X.Y[.Z]` merge forms of `qrun-maven-prepare.mjs` (`RELEASE_BRANCH_GREP`, which replaces the legacy
`Merge.*release.*into.*develop` forms) and the unchanged third `Bump.*version.*after.*release.*v` of `calculate_version.sh`. Git matches per message line (REG_NEWLINE), so each line is searched alone;
`[:alnum:]` is read as ASCII."""

DEVELOP_NEXT_MINOR: Final = "develop-next-minor"
RC_FROM_BRANCH: Final = "rc-from-branch"
PATCH_BUMP: Final = "patch-bump"
STABLE_FROM_TAG: Final = "stable-from-tag"
FEATURE_SNAPSHOT: Final = "feature-snapshot"
EFFECTIVE_VERSION_RULES: Final = (DEVELOP_NEXT_MINOR, RC_FROM_BRANCH, PATCH_BUMP, STABLE_FROM_TAG, FEATURE_SNAPSHOT)
"""The closed vocabulary a channel's `effective_version.rule` may use (a rule that cannot run is still named, so a repository states its need and the event fails visibly)."""
RC_BASELINE_NONE: Final = "none"
RC_BASELINE_RECONCILED: Final = "reconciled"
RC_BASELINES: Final = (RC_BASELINE_NONE, RC_BASELINE_RECONCILED)
DEFAULT_VERSION_PROPERTY: Final = "revision"
VERSION_PROPERTY_PATTERN: Final = re.compile(r"[A-Za-z][A-Za-z0-9._-]*")

_RC = re.compile(r"(\d+)\.(\d+)\.(\d+)-RC\.(\d+)")
_SNAPSHOT = re.compile(r"(\d+)\.(\d+)\.(\d+)-SNAPSHOT")
_FEATURE_SNAPSHOT = re.compile(r"(\d+)\.(\d+)\.(\d+)-[a-zA-Z0-9-]+-[a-f0-9]+-SNAPSHOT")
_STABLE = re.compile(r"(\d+)\.(\d+)\.(\d+)")
_RELEASE_BRANCH = re.compile(r"release/([0-9]+)\.([0-9]+)")
_STABLE_TAG = re.compile(r"v(\d+\.\d+\.\d+)")
_FIRST_RC_LINE = re.compile(r"\d+\.\d+\.0-SNAPSHOT")


def version_parts(version: str) -> tuple[int, int, int]:
   """Return (major, minor, patch) of a version in one of the captured formats; anything else is an error, never a guess."""
   candidate = version[1:] if version.startswith("v") else version
   for pattern in (_RC, _SNAPSHOT, _FEATURE_SNAPSHOT, _STABLE):
      matched = pattern.fullmatch(candidate)
      if matched is not None:
         return int(matched.group(1)), int(matched.group(2)), int(matched.group(3))
   raise PipelineError(f"the declared version {version!r} is not a RC, SNAPSHOT, feature SNAPSHOT or stable X.Y.Z version, so no effective version can be calculated")


def release_activity_since(build_started: int) -> int:
   """Return the earliest committer time (epoch seconds) inside the window that ends at the build start, as the captured `QRUN_VERSION_SINCE=@<started - 3 days>`."""
   return build_started - RELEASE_ACTIVITY_WINDOW_SECONDS


def message_matches_release_activity(message: str) -> bool:
   """Return whether any line of a commit message matches one of the captured release-activity predicates."""
   return any(pattern.search(line) is not None for line in message.split("\n") for pattern in RELEASE_ACTIVITY_PATTERNS)


def has_release_activity(commits: Iterable[tuple[int, str]], since: int) -> bool:
   """Return whether any (committer epoch, message) commit is at or after `since` and matches the release-activity predicate."""
   return any(committed >= since and message_matches_release_activity(message) for committed, message in commits)


def _develop_next_minor(declared: str, branch: str, release_activity: Callable[[], bool] | None) -> str:
   """Return the develop version: RC and stable bump to the next minor SNAPSHOT, a declared SNAPSHOT is kept unless recent release activity bumps it."""
   major, minor, _ = version_parts(declared)
   if branch != "develop":
      raise PipelineError(f"the {DEVELOP_NEXT_MINOR} rule applies to branch develop, not {branch!r}")
   if _RC.fullmatch(declared) or _STABLE.fullmatch(declared):
      return f"{major}.{minor + 1}.0-SNAPSHOT"
   if _SNAPSHOT.fullmatch(declared) is None and _FEATURE_SNAPSHOT.fullmatch(declared) is None:
      raise PipelineError(f"develop declares {declared!r}, which is not an RC, SNAPSHOT or stable version")
   if release_activity is None:
      raise ReleaseActivityError(f"develop declares {declared!r}: the captured calculator bumps a SNAPSHOT when a release was merged into develop within 3 days of the build start, and no release-activity input (the server's prepare job start plus the commit history) was supplied, so the next version is unknown")
   return f"{major}.{minor + 1}.0-SNAPSHOT" if release_activity() else declared


def _rc_from_branch(declared: str, branch: str, rc_baseline: str) -> str:
   """Return the next release candidate of a `release/X.Y` branch: one more than a declared RC counter, otherwise the first."""
   matched = _RELEASE_BRANCH.search(branch) if branch.startswith("release/") else None
   if matched is None:
      raise PipelineError(f"branch {branch!r} is not release/X.Y, so no release candidate version can be calculated")
   if rc_baseline == RC_BASELINE_RECONCILED and _FIRST_RC_LINE.fullmatch(declared):
      raise PipelineError(
         f"release branch declares {declared!r} under rc_baseline reconciled: the captured calculator restores a tagged first-parent RC counter after a reconciling merge, which needs the "
         "commit's parents, their root POM revisions and tag ancestry; those are not inputs this contract carries, so the next counter is unknown"
      )
   current = _RC.fullmatch(declared)
   number = int(current.group(4)) + 1 if current is not None else 1
   return f"{matched.group(1)}.{matched.group(2)}.0-RC.{number}"


def effective_version(rule: str, declared: str, branch: str, tag: str, rc_baseline: str = RC_BASELINE_NONE, release_activity: Callable[[], bool] | None = None) -> str:  # noqa: PLR0913, PLR0917 - the established signature of every caller
   """Return the version the build must use for this rule, declared version and the event's branch or tag; refuse where a required input is not carried.

   `release_activity` answers the develop SNAPSHOT question (see the module text); it is called only when the answer decides the version, at most once.
   """
   if rule not in EFFECTIVE_VERSION_RULES:
      raise PipelineError(f"unsupported effective version rule {rule!r}; supported: {', '.join(EFFECTIVE_VERSION_RULES)}")
   major, minor, patch = version_parts(declared)
   if rule == FEATURE_SNAPSHOT:
      raise PipelineError("the feature-snapshot version needs the originating feature branch of the publish tag (git branch -r --contains HEAD) and the commit abbreviation as git prints it; neither is an input this contract carries, so no feature version is calculated")
   if rule == STABLE_FROM_TAG:
      if not tag:
         raise PipelineError(f"the {STABLE_FROM_TAG} rule needs a pushed tag")
      matched = _STABLE_TAG.fullmatch(tag)
      if matched is None:
         raise PipelineError(f"tag {tag!r} is not vX.Y.Z, so no stable version can be calculated (the release-candidate tag route is not carried)")
      return matched.group(1)
   if tag or not branch:
      raise PipelineError(f"the {rule} rule needs a pushed branch, not a tag or a missing branch")
   if rule == DEVELOP_NEXT_MINOR:
      return _develop_next_minor(declared, branch, release_activity)
   if rule == RC_FROM_BRANCH:
      return _rc_from_branch(declared, branch, rc_baseline)
   if not branch.startswith("hotfix/"):
      raise PipelineError(f"the {PATCH_BUMP} rule applies to hotfix/ branches, not {branch!r}")
   return f"{major}.{minor}.{patch + 1}"


def with_version_property(commands: Sequence[Sequence[str]], property_name: str, version: str) -> list[list[str]]:
   """Return the native commands with `-D<property>=<version>` appended to each; a command that already sets the property is a conflict."""
   argument = f"-D{property_name}={version}"
   result: list[list[str]] = []
   for command in commands:
      if any(token == f"-D{property_name}" or token.startswith(f"-D{property_name}=") for token in command):
         raise PipelineError(f"a declared command already sets -D{property_name}; the effective version strategy owns that property")
      result.append([*command, argument])
   return result