On-call notification audit

apiVersion: workflows.ext.grafana.app/v1alpha1
kind: Definition
metadata:
  name: on-call-notification-audit
  annotations:
    grafana.app/folder: <FOLDER_UID>
spec:
  name: On-call notification audit
  enabled: false
  startWhen:
    matchingRules: []
    schedules:
      - 'CRON_TZ=America/New_York 0 9 * * 1'
  resources:
    secrets:
      irmApiToken: 'secret://irm-api-token'
  steps:
    - id: fetch-users
      type: http.call
      name: Fetch users
      description: >-
        Reads every user's phone verification from the OnCall API. Replace your-oncall-api-url
        with the API URL shown in IRM under Settings. Also create a workflow secret named
        irm-api-token that holds an OnCall API token, created in IRM under Settings. Service
        account tokens don't work, because the notification rules endpoint doesn't accept them.
      inputs:
        method: 'GET'
        url: 'https://your-oncall-api-url/api/v1/users/?perpage=100'
        authMethod: 'bearer'
        authSecret: '${resources.secret.irmApiToken}'
    - id: fetch-important-rules
      type: http.call
      name: Fetch important notification rules
      description: Reads everyone's important notification rules from the same API. Use the same API URL as Fetch users.
      inputs:
        method: 'GET'
        url: 'https://your-oncall-api-url/api/v1/personal_notification_rules/?important=true&perpage=100'
        authMethod: 'bearer'
        authSecret: '${resources.secret.irmApiToken}'
    - id: check-responses
      type: filter
      name: Fail if the API calls failed
      description: Fails the run with an error instead of posting a wrong report when either API call does not succeed. No changes needed.
      inputs:
        condition: "steps['fetch-users'].outputs.responseCode == 200 && steps['fetch-important-rules'].outputs.responseCode == 200"
        failOnExit: 'true'
    - id: audit
      type: code.execute
      name: Find notification gaps
      description: >-
        Checks every user and builds the Slack Block Kit report, mentioning people whose Slack
        account is linked. reachableRuleTypes in args lists the rule types that count as reaching
        someone. Add notify_by_sms to count text messages. skipUsernamePattern skips Grafana
        service accounts, whose logins start with sa-.
      inputs:
        language: 'python'
        args: '{"reachableRuleTypes": ["notify_by_phone_call", "notify_by_mobile_app", "notify_by_mobile_app_critical"], "skipUsernamePattern": "^sa-[0-9]+-"}'
        code: |
          import json
          import re
          from datetime import datetime, timezone

          users_page = json.loads(context["steps"]["fetch-users"]["outputs"]["responseBody"])
          rules_page = json.loads(context["steps"]["fetch-important-rules"]["outputs"]["responseBody"])
          all_users = users_page.get("results", [])
          rules = rules_page.get("results", [])

          # Grafana service accounts show up as users, with logins like sa-1-name. They are never on call.
          skip = re.compile(args.get("skipUsernamePattern") or "(?!)")
          users = [u for u in all_users if not skip.search(u.get("username") or "")]
          skipped = len(all_users) - len(users)
          reachable = set(args.get("reachableRuleTypes", []))

          rule_types = {}
          for rule in rules:
              rule_types.setdefault(rule.get("user_id"), set()).add(rule.get("type"))

          def who(user):
              slack = user.get("slack") or {}
              if slack.get("user_id"):
                  return f"<@{slack['user_id']}>"
              return user.get("username") or user.get("email") or user.get("id")

          no_verified_phone, not_reachable, calls_unverified_phone = [], [], []
          for user in users:
              types = rule_types.get(user.get("id"), set())
              verified = bool(user.get("is_phone_number_verified"))
              if not verified:
                  no_verified_phone.append(who(user))
              if not types & reachable:
                  not_reachable.append(who(user))
              if "notify_by_phone_call" in types and not verified:
                  calls_unverified_phone.append(who(user))
          today = datetime.now(timezone.utc)
          flagged = set(no_verified_phone) | set(not_reachable) | set(calls_unverified_phone)
          covered = len(users) - len(flagged)

          def names(people, limit=40):
              shown = ", ".join(sorted(people)[:limit])
              return shown + (f" and {len(people) - limit} more" if len(people) > limit else "")

          def gap(title, explanation, people, ok_text):
              if not people:
                  text = f":white_check_mark: *{ok_text}*"
              else:
                  text = f":warning: *{title}* ({len(people)})\n_{explanation}_\n{names(people)}"
              return {"type": "section", "text": {"type": "mrkdwn", "text": text}}

          notes = []
          if users_page.get("count", len(all_users)) > len(all_users):
              notes.append(f":warning: Read {len(all_users)} of {users_page['count']} users. Add a team_id filter to the Fetch users URL to audit one team at a time.")
          if rules_page.get("count", len(rules)) > len(rules):
              notes.append(f":warning: Read {len(rules)} of {rules_page['count']} important rules, so some people above may be listed by mistake.")
          else:
              notes.append(f"Read all {len(rules)} important notification rules.")
          notes.append("Mobile push means the rules use the IRM mobile app. It doesn't confirm a device is paired.")

          blocks = [
              {"type": "header", "text": {"type": "plain_text", "text": ":rotating_light: On-call notification audit"}},
              {"type": "context", "elements": [{"type": "mrkdwn", "text": f"{today:%A, %B} {today.day} · {len(users)} people checked" + (f" · {skipped} service account{'' if skipped == 1 else 's'} skipped" if skipped else "")}]},
              {"type": "section", "fields": [
                  {"type": "mrkdwn", "text": f"*Fully covered*\n{covered} of {len(users)}"},
                  {"type": "mrkdwn", "text": f"*No verified phone*\n{len(no_verified_phone)}"},
                  {"type": "mrkdwn", "text": f"*Can't be reached on important pages*\n{len(not_reachable)}"},
                  {"type": "mrkdwn", "text": f"*Calls an unverified phone*\n{len(calls_unverified_phone)}"},
              ]},
              {"type": "divider"},
              gap("No verified phone number", "Add and verify a phone number in IRM user settings.",
                  no_verified_phone, "Everyone has a verified phone number"),
              gap("Important rules have no phone call or mobile push", "Important pages may never reach these people.",
                  not_reachable, "Everyone's important rules use a phone call or mobile push"),
              gap("Important rules call a phone that isn't verified", "These calls can't be delivered until the number is verified.",
                  calls_unverified_phone, "No important rules call an unverified phone"),
              {"type": "divider"},
              {"type": "context", "elements": [{"type": "mrkdwn", "text": note} for note in notes]},
          ]

          summary = (f"On-call notification audit: {covered} of {len(users)} people fully covered, "
                     f"{len(flagged)} need attention.")
          print(json.dumps({"summary": summary, "blocks": blocks}))
    - id: post-report
      type: slack.message.post
      name: Post the audit to Slack
      description: Posts the audit as Block Kit, with a plain-text summary for notifications. Set channelID to the channel where your on-call leads look.
      inputs:
        channelID: ''
        messageText: "${json_parse(steps.audit.outputs.stdout).summary}"
        blocks: "${json_stringify(json_parse(steps.audit.outputs.stdout).blocks)}"