#!/usr/bin/python3
"""Cancel superseded, never-started CI runs in "latest:" and "replace:" concurrency groups.

Runs inside the forgejo LXD container on ollivanders (systemd timer, every minute).

Workflows opt in with
    concurrency:
      group: latest:${{ github.workflow }}@${{ github.ref }}
      cancel-in-progress: ${{ github.ref != 'refs/heads/main' }}   # default branch: queue
On the default branch that makes Forgejo QUEUE each new run behind the previous one
(concurrency_type 1 = QueueBehind) instead of killing it, so a run that has started is
never cancelled by a later push. Forgejo would then run every commit in order; this
sweeper keeps that bounded, like GitHub's "one running + one pending":

  per (repo, group), among runs that have not finished:
    - never touch a run that has started: any job claimed by a runner or finished. (The run
      status alone is not enough: a run whose first jobs finished while later ones wait for
      a runner is back to status "waiting".)
    - keep the NEWEST unstarted run (it tests the latest commit)
    - if nothing in the group has started, also keep the OLDEST unstarted run: it is the
      one eligible for a runner right now and holds the group's place in Forgejo's
      oldest-first queue, so constant pushes can't keep sending main to the back
    - never cancel a run whose commit title starts with "Release " (version bump: it
      publishes the agent artifacts)
    - cancel every other unstarted run (POST /actions/runs/{id}/cancel)

Scheduled heavy workflows (mutation / chaos nightlies on the colo lane) use
    concurrency:
      group: replace:${{ github.workflow }}@${{ github.ref }}
      cancel-in-progress: false
and per (repo, group) keep only the NEWEST unstarted run: a newer queued nightly replaces an
older queued one (both would test nearly the same code, and each holds a colo slot for up to
3 h), while a started run is still never touched. (2026-10-18: the colo lane queued 10+ h with
agents re-dispatching nightlies that were already waiting.)

Reads forgejodb (credentials from /etc/forgejo/app.ini); cancels only through the API
with the token in /etc/ci-supersede-sweeper/token (scope write:repository).
--dry-run prints what it would cancel.
"""
import configparser, json, re, subprocess, sys, urllib.request

APP_INI = "/etc/forgejo/app.ini"
TOKEN_FILE = "/etc/ci-supersede-sweeper/token"
API = "http://127.0.0.1:3000/api/v1"
WAITING, RUNNING, BLOCKED = 5, 6, 7
QUEUE_BEHIND = 1

SQL = f"""
select r.id, r.repo_id, r.concurrency_group, r.index,
       case when exists (select 1 from action_run_job j where j.run_id = r.id
                         and (j.task_id <> 0 or j.status in (1, 2, {RUNNING})))
            then 1 else 0 end as started,
       r.commit_sha, u.lower_name, repo.lower_name,
       (r.title ilike 'release %') as is_release
from action_run r join repository repo on repo.id = r.repo_id join "user" u on u.id = repo.owner_id
where (r.concurrency_group like 'latest:%' or r.concurrency_group like 'replace:%')
  and r.concurrency_type = {QUEUE_BEHIND}
  and r.status in ({WAITING}, {RUNNING}, {BLOCKED})
order by r.repo_id, r.concurrency_group, r.index
"""


def db_rows():
    ini = configparser.ConfigParser(interpolation=None, strict=False)
    with open(APP_INI) as f:
        ini.read_string("[root]\n" + f.read())
    db = ini["database"]
    host, _, port = db["HOST"].partition(":")
    pw = db.get("PASSWD", "").strip().strip("`")
    out = subprocess.run(
        ["psql", "-h", host, "-p", port or "5432", "-U", db["USER"], "-d", db["NAME"],
         "-X", "-A", "-t", "-F", "\t", "-c", SQL],
        env={"PGPASSWORD": pw, "PATH": "/usr/bin:/bin"}, capture_output=True, text=True, timeout=60, check=True,
    ).stdout
    rows = []
    for line in out.splitlines():
        if not line.strip():
            continue
        rid, repo_id, group, idx, started, sha, owner, name, rel = line.split("\t")
        rows.append(dict(id=int(rid), repo_id=int(repo_id), group=group, index=int(idx),
                         started=started == "1", sha=sha[:10], owner=owner, repo=name, release=rel == "t"))
    return rows


def lineage(group):
    """A group may carry a ':<40-hex sha>' suffix (default-branch push workflows use one group per
    commit so Forgejo does not serialise runs; 2026-11-06). The sweeper treats all of them as one
    lineage, so it still keeps only started runs + the newest unstarted one."""
    return re.sub(r":[0-9a-f]{40}$", "", group)


def plan(rows):
    groups = {}
    for r in rows:
        groups.setdefault((r["repo_id"], lineage(r["group"])), []).append(r)
    cancel = []
    for runs in groups.values():
        runs.sort(key=lambda r: r["index"])
        started = [r for r in runs if r["started"]]
        unstarted = [r for r in runs if not r["started"]]
        if len(unstarted) < 2:
            continue
        keep = {unstarted[-1]["id"]}
        if not started and not runs[0]["group"].startswith("replace:"):
            keep.add(unstarted[0]["id"])
        # A version-bump commit ("Release x.y.z: ...") must never lose its run: that run
        # builds, signs and publishes the version's agent artifacts, and a cancelled one
        # leaves devices on some platforms unable to self-update (0.1.297, 0.1.299).
        keep |= {r["id"] for r in unstarted if r.get("release")}
        cancel += [r for r in unstarted if r["id"] not in keep]
    return cancel


# ---- priority + fair share (2026-11-06, owner: "do not defer work to a clock; manage contention with priority") ----
# Forgejo hands jobs to a runner by action_run.priority DESC, then job `updated` ASC. Nothing is delayed to a time window:
# every run starts as soon as a slot is free; this only decides who goes FIRST.
#   100  deploy-critical: Rustinion ci-cd push, deploy-latest-green, release*/tag runs, security/CVE fixes
#     0  ordinary push / PR / dispatch CI, minus fair share: -2 per run of the SAME repo already running (cap -10),
#        so a repo (or one burst of agent pushes) cannot occupy every slot ahead of other repos
#   -20  heavy optional: live-iso*, binhost, nightly*, mutation*, ci-base, media benches, gentoo-ebuilds, schedule events
PRIO_SQL = r"""
with cls as (
  select r.id, r.priority cur, r.repo_id,
    case
      when (repo.lower_name = 'rustinion' and r.workflow_id = 'ci-cd.yml' and r.trigger_event = 'push')
        or r.workflow_id = 'deploy-latest-green.yml' or r.workflow_id ilike 'release%' or r.ref like 'refs/tags/%'
        or r.title ~* '^release |security|cve-' then 100
      when r.workflow_id ~* '^(live-iso|binhost|nightly|mutation|build-ci-base|ra-media|ra-audio|ra-codec|gentoo-ebuilds)'
        or r.trigger_event = 'schedule' then -20
      else 0 end base
  from action_run r join repository repo on repo.id = r.repo_id
  where r.status in (5, 6, 7)),
run as (select repo_id, count(*) n from action_run where status = 6 group by repo_id),
want as (select c.id, c.cur, c.base - case when c.base = 0 then least(10, 2 * coalesce(run.n, 0)::int) else 0 end p
         from cls c left join run on run.repo_id = c.repo_id)
@@SQL@@
"""
PRIO_UPDATE = "update action_run r set priority = w.p from want w where r.id = w.id and r.priority <> w.p"
PRIO_COUNT = "select count(*) from want where cur <> p"


def psql(sql):
    ini = configparser.ConfigParser(interpolation=None, strict=False)
    with open(APP_INI) as f:
        ini.read_string("[root]\n" + f.read())
    db = ini["database"]
    host, _, port = db["HOST"].partition(":")
    pw = db.get("PASSWD", "").strip().strip("`")
    return subprocess.run(
        ["psql", "-h", host, "-p", port or "5432", "-U", db["USER"], "-d", db["NAME"], "-X", "-A", "-t", "-c", sql],
        env={"PGPASSWORD": pw, "PATH": "/usr/bin:/bin"}, capture_output=True, text=True, timeout=60, check=True).stdout


def set_priorities(dry):
    out = psql(PRIO_SQL.replace("@@SQL@@", PRIO_COUNT if dry else PRIO_UPDATE)).strip()
    if out and out != "0":
        print(("would change" if dry else "priority"), out)


def main():
    dry = "--dry-run" in sys.argv
    try:
        set_priorities(dry)
    except Exception as e:  # never let priorities block the cancel pass
        print("priority pass failed", e, file=sys.stderr)
    cancel = plan(db_rows())
    if not cancel:
        return
    token = None if dry else open(TOKEN_FILE).read().strip()
    for r in cancel:
        what = f"{r['owner']}/{r['repo']} run {r['id']} (#{r['index']}, {r['sha']}) in {r['group']}"
        if dry:
            print("would cancel", what)
            continue
        req = urllib.request.Request(f"{API}/repos/{r['owner']}/{r['repo']}/actions/runs/{r['id']}/cancel",
                                     method="POST", headers={"Authorization": f"token {token}"})
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                print("cancelled superseded", what, resp.status)
        except Exception as e:  # keep going; next tick retries
            print("FAILED to cancel", what, e, file=sys.stderr)


if __name__ == "__main__":
    main()
