SharedWorkflows/.gitea/workflows/notifications.yml

name: Notification

on:
  workflow_call:
    inputs:
      job_status:
        required: true
        type: string
      job_name:
        required: true
        type: string
      notify_on_release:
        required: false
        type: boolean
        default: true
    secrets:
      NTFY_TOPIC: { required: true }
      NTFY_TOKEN: { required: true }
      NTFY_SERVER: { required: true }
      API_GITEA_TOKEN: { required: true }

jobs:
  notify:
    runs-on: docker
    steps:
      - name: Detect previous workflow state
        id: previous
        shell: bash
        env:
          API_GITEA_TOKEN: ${{ secrets.API_GITEA_TOKEN }}
        run: |
          set -euo pipefail

          python3 <<'PY'
          import json
          import os
          import urllib.request

          server = "${{ gitea.server_url }}"
          repo = "${{ gitea.repository }}"
          current_run_number = int("${{ gitea.run_number }}")
          current_branch = "${{ gitea.ref_name }}"
          current_event = "${{ gitea.event_name }}"
          token = os.environ["API_GITEA_TOKEN"]

          url = (
              f"{server}/api/v1/repos/{repo}/actions/runs"
              f"?page=1&limit=5"
          )

          print(f"Fetching workflow runs from: {url}")
          print(f"Current run number: {current_run_number}")
          print(f"Current branch: {current_branch}")
          print(f"Current event: {current_event}")

          req = urllib.request.Request(
              url,
              headers={
                  "Authorization": f"token {token}",
                  "Accept": "application/json",
              },
          )

          with urllib.request.urlopen(req) as response:
              data = json.loads(response.read().decode("utf-8"))

          runs = data.get("workflow_runs", [])

          print(f"Received {len(runs)} workflow runs")

          previous = "unknown"

          for run in runs:
              run_number = run.get("run_number")
              status = run.get("status")
              conclusion = run.get("conclusion")
              branch = run.get("head_branch")
              event = run.get("event")

              print(
                  f"Inspecting run #{run_number}: "
                  f"status={status}, "
                  f"conclusion={conclusion}, "
                  f"branch={branch}, "
                  f"event={event}"
              )

              # aktuellen Run überspringen
              if int(run_number) == current_run_number:
                  print(" -> skipping current run")
                  continue

              # nur abgeschlossene Runs
              if status != "completed":
                  print(" -> skipping non-completed run")
                  continue

              # nur gleicher Branch
              if branch != current_branch:
                  print(" -> skipping different branch")
                  continue

              # nur gleiches Event
              if event != current_event:
                  print(" -> skipping different event")
                  continue

              previous = conclusion or "unknown"

              print(f" -> selected previous conclusion: {previous}")
              break

          print(f"Previous conclusion final: {previous}")

          with open(os.environ["GITEA_OUTPUT"], "a", encoding="utf-8") as f:
              f.write(f"previous_conclusion={previous}\n")
          PY

      - name: Send Notification
        shell: bash
        run: |
          set -euo pipefail

          CURRENT_STATUS="${{ inputs.job_status }}"
          PREVIOUS_STATUS="${{ steps.previous.outputs.previous_conclusion }}"
          IS_RELEASE="false"

          if [[ "${{ gitea.ref }}" == refs/tags/v* ]]; then
            IS_RELEASE="true"
          fi

          echo "Current status: ${CURRENT_STATUS}"
          echo "Previous status: ${PREVIOUS_STATUS}"
          echo "Is release: ${IS_RELEASE}"

          if [[ "$IS_RELEASE" == "true" && "$CURRENT_STATUS" == "success" ]]; then
            ICON="🚀"
            TITLE="New release available: ${{ gitea.ref_name }} for ${{ gitea.repository }}"
          elif [[ "$CURRENT_STATUS" == "failure" ]]; then
            ICON="❌"
            TITLE="Failed: ${{ inputs.job_name }} for ${{ gitea.repository }}"
          elif [[ "$CURRENT_STATUS" == "success" && "$PREVIOUS_STATUS" == "failure" ]]; then
            ICON="✅"
            TITLE="Healed: ${{ inputs.job_name }} for ${{ gitea.repository }}"
          else
            echo "No notification needed."
            exit 0
          fi

          COMMIT_SUBJECT="$(curl -fsS \
            -H "Authorization: Bearer ${{ secrets.API_GITEA_TOKEN }}" \
            "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/git/commits/${{ gitea.sha }}" \
            | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('RepoCommit',{}).get('message','').split('\n')[0])" \
            2>/dev/null || echo 'Commit info unavailable')"
          RUN_URL="${{ gitea.server_url }}/${{ gitea.repository }}/actions/runs/${{ gitea.run_number }}"

          cat <<EOF >/tmp/ntfy-payload.json
          {
            "topic": "${{ secrets.NTFY_TOPIC }}",
            "title": "${ICON} ${TITLE}",
            "message": "Ref: ${{ gitea.ref_name }}\nStatus: ${CURRENT_STATUS}\nPrevious: ${PREVIOUS_STATUS}\nCommit: ${COMMIT_SUBJECT}\n\nRun URL: ${RUN_URL}",
            "click": "${RUN_URL}",
            "actions": [
              { "action": "view", "label": "Open Run", "url": "${RUN_URL}" }
            ]
          }
          EOF

          curl -fsS \
            -H "Authorization: Bearer ${{ secrets.NTFY_TOKEN }}" \
            -H "Content-Type: application/json" \
            -d @/tmp/ntfy-payload.json \
            "${{ secrets.NTFY_SERVER }}"