Listing and Acknowledging Alerts with the AI DBA Workbench API

# httpx, an HTTP client with a requests-like interface.
httpx>=0.28,<1

{
  "name": "alerts",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/main.js"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "typescript": "^5.9.0"
  }
}

<dependencies>
  <!-- Jackson, for JSON; the HTTP client is part of the JDK. -->
  <dependency>
    <groupId>tools.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>3.2.4</version>
  </dependency>
</dependencies>

<!-- The program uses only HttpClient and System.Text.Json, which are
     part of .NET, so it needs no packages. -->

// The program uses only the Go standard library, so it needs no modules.
go 1.25.0

[dependencies]
# The blocking client of reqwest, with rustls for TLS and JSON bodies.
reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] }
serde_json = "1"

source "https://rubygems.org"

# The program uses only Net::HTTP and JSON, which come with Ruby, so it
# needs no gems.

import os
import sys
import time

import httpx

// The program needs no imports: fetch and process are built in to Node.js.

import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpTimeoutException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import tools.jackson.core.JacksonException;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;

using System.Diagnostics;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;
using System.Text.Json.Nodes;

import (
    "bytes"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "net/url"
    "os"
    "strings"
    "time"
)

use std::env;
use std::error::Error;
use std::process::ExitCode;
use std::sync::LazyLock;
use std::thread;
use std::time::{Duration, Instant};

use reqwest::blocking::Client;
use reqwest::{Method, Url};
use serde_json::{json, Value};

require "json"
require "net/http"

# Stop at the first failing command, treat an unset variable as an error,
# and fail a pipeline when any command in it fails.
set -euo pipefail

# The script needs curl to make the requests and jq to read the replies.
for tool in curl jq; do
    command -v "$tool" >/dev/null || { echo "$tool is required" >&2; exit 1; }
done

# The rule that the program makes fire. The metric_staleness rule alerts when
# a probe's data is out of date, and the alerter checks it every minute.
RULE_NAME = "metric_staleness"

# The longest the whole run waits for alerts, in seconds. The alerter
# evaluates its rules once a minute and clears an alert up to 30 seconds
# after it fires, so this allows for waiting out the alert from an earlier
# run, one alert that clears before the program can acknowledge it, and a
# second evaluation.
WAIT_SECONDS = 150

# How often the program polls the alerts, in seconds.
POLL_SECONDS = 0.5


def settings() -> dict:
    """Reads the settings from the environment.

    The or operator treats an empty variable as unset.
    """
    token = os.environ.get("WORKBENCH_TOKEN") or ""
    connection_id = os.environ.get("WORKBENCH_CONNECTION_ID") or ""
    if not (token and connection_id):
        raise RuntimeError(
            "WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set"
        )
    return {
        "base_url": (
            os.environ.get("WORKBENCH_URL") or "http://localhost:8080"
        ).rstrip("/"),
        "token": token,
        "connection_id": connection_id,
    }

// The rule that the program makes fire. The metric_staleness rule alerts when
// a probe's data is out of date, and the alerter checks it every minute.
const RULE_NAME = "metric_staleness";

// The longest the whole run waits for alerts, in milliseconds. The alerter
// evaluates its rules once a minute and clears an alert up to 30 seconds
// after it fires, so this allows for waiting out the alert from an earlier
// run, one alert that clears before the program can acknowledge it, and a
// second evaluation.
const WAIT_MS = 150_000;

// How often the program polls the alerts, in milliseconds.
const POLL_MS = 500;

interface Settings {
  baseUrl: string;
  token: string;
  connectionId: string;
}

// Read the settings from the environment. The || operator, unlike ??, also
// treats an empty variable as unset.
function settings(): Settings {
  const token = process.env.WORKBENCH_TOKEN || "";
  const connectionId = process.env.WORKBENCH_CONNECTION_ID || "";
  if (!token || !connectionId) {
    throw new Error("WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set");
  }
  const baseUrl = process.env.WORKBENCH_URL || "http://localhost:8080";
  return { baseUrl: baseUrl.replace(/\/+$/, ""), token, connectionId };
}

/**
 * The rule that the program makes fire. The metric_staleness rule
 * alerts when a probe's data is out of date, and the alerter checks it
 * every minute.
 */
static final String RULE_NAME = "metric_staleness";

/**
 * The longest the whole run waits for alerts. The alerter evaluates its
 * rules once a minute and clears an alert up to 30 seconds after it
 * fires, so this allows for waiting out the alert from an earlier run,
 * one alert that clears before the program can acknowledge it, and a
 * second evaluation.
 */
static final Duration WAIT_TIME = Duration.ofSeconds(150);

/** How often the program polls the alerts. */
static final Duration POLL_INTERVAL = Duration.ofMillis(500);

/** The settings that the program reads from the environment. */
record Settings(String baseUrl, String token, String connectionId) {}

/**
 * Returns the value of an environment variable, or the fallback if the
 * variable is unset or empty, as an unset variable would be.
 */
static String envOr(String name, String fallback) {
    String value = System.getenv(name);
    return (value != null && !value.isEmpty()) ? value : fallback;
}

/** Reads the settings from the environment. */
static Settings settings() throws IOException {
    String token = envOr("WORKBENCH_TOKEN", "");
    String connectionId = envOr("WORKBENCH_CONNECTION_ID", "");
    if (token.isEmpty() || connectionId.isEmpty()) {
        throw new IOException(
                "WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set");
    }
    String baseUrl = envOr("WORKBENCH_URL", "http://localhost:8080")
            .replaceAll("/+$", "");
    return new Settings(baseUrl, token, connectionId);
}

// The rule that the program makes fire. The metric_staleness rule alerts
// when a probe's data is out of date, and the alerter checks it every
// minute.
const string RuleName = "metric_staleness";

// The longest the whole run waits for alerts. The alerter evaluates its
// rules once a minute and clears an alert up to 30 seconds after it fires,
// so this allows for waiting out the alert from an earlier run, one alert
// that clears before the program can acknowledge it, and a second
// evaluation.
var waitTime = TimeSpan.FromSeconds(150);

// How often the program polls the alerts.
var pollInterval = TimeSpan.FromMilliseconds(500);

// Returns the value of an environment variable, or null if the variable
// is unset or empty, as an unset variable would be.
static string? EnvValue(string name) =>
    Environment.GetEnvironmentVariable(name) is { Length: > 0 } value
        ? value
        : null;

// Reads the settings from the environment.
static (string BaseUrl, string Token, string ConnectionId) Settings()
{
    var token = EnvValue("WORKBENCH_TOKEN");
    var connectionId = EnvValue("WORKBENCH_CONNECTION_ID");
    if (token is null || connectionId is null)
    {
        throw new InvalidOperationException(
            "WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set");
    }
    var baseUrl = EnvValue("WORKBENCH_URL") ?? "http://localhost:8080";
    return (baseUrl.TrimEnd('/'), token, connectionId);
}

// ruleName is the rule that the program makes fire. The metric_staleness
// rule alerts when a probe's data is out of date, and the alerter checks it
// every minute.
const ruleName = "metric_staleness"

// waitTime is the longest the whole run waits for alerts. The alerter
// evaluates its rules once a minute and clears an alert up to 30 seconds
// after it fires, so this allows for waiting out the alert from an earlier
// run, one alert that clears before the program can acknowledge it, and a
// second evaluation.
const waitTime = 150 * time.Second

// pollTime is how often the program polls the alerts.
const pollTime = 500 * time.Millisecond

// settings holds the values that the program reads from the environment.
type settings struct {
    baseURL      string
    token        string
    connectionID string
}

// envOr returns the value of an environment variable, or the fallback if
// the variable is unset or empty, as an unset variable would be.
func envOr(name, fallback string) string {
    if value := os.Getenv(name); value != "" {
        return value
    }
    return fallback
}

// loadSettings reads the settings from the environment.
func loadSettings() (settings, error) {
    s := settings{
        baseURL:      strings.TrimRight(envOr("WORKBENCH_URL", "http://localhost:8080"), "/"),
        token:        os.Getenv("WORKBENCH_TOKEN"),
        connectionID: os.Getenv("WORKBENCH_CONNECTION_ID"),
    }
    if s.token == "" || s.connectionID == "" {
        return s, errors.New("WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set")
    }
    return s, nil
}

/// The rule that the program makes fire. The metric_staleness rule alerts
/// when a probe's data is out of date, and the alerter checks it every
/// minute.
const RULE_NAME: &str = "metric_staleness";

/// The longest the whole run waits for alerts. The alerter evaluates its
/// rules once a minute and clears an alert up to 30 seconds after it
/// fires, so this allows for waiting out the alert from an earlier run,
/// one alert that clears before the program can acknowledge it, and a
/// second evaluation.
const WAIT_TIME: Duration = Duration::from_secs(150);

/// How often the program polls the alerts.
const POLL_INTERVAL: Duration = Duration::from_millis(500);

/// Any error, boxed, so that a function can return errors of several types
/// with the ? operator.
type Result<T> = std::result::Result<T, Box<dyn Error>>;

/// The settings that the program reads from the environment.
struct Settings {
    base_url: String,
    token: String,
    connection_id: String,
}

/// Returns the value of an environment variable, if it is set and not
/// empty; an empty variable counts as unset.
fn env_value(name: &str) -> Option<String> {
    env::var(name).ok().filter(|value| !value.is_empty())
}

/// Reads the settings from the environment.
fn settings() -> Result<Settings> {
    let (Some(token), Some(connection_id)) = (
        env_value("WORKBENCH_TOKEN"),
        env_value("WORKBENCH_CONNECTION_ID"),
    ) else {
        return Err("WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set".into());
    };
    let base_url = env_value("WORKBENCH_URL")
        .unwrap_or_else(|| "http://localhost:8080".to_string())
        .trim_end_matches('/')
        .to_string();
    Ok(Settings {
        base_url,
        token,
        connection_id,
    })
}

# The rule that the program makes fire. The metric_staleness rule alerts
# when a probe's data is out of date, and the alerter checks it every
# minute.
RULE_NAME = "metric_staleness"

# The longest the whole run waits for alerts, in seconds. The alerter
# evaluates its rules once a minute and clears an alert up to 30 seconds
# after it fires, so this allows for waiting out the alert from an earlier
# run, one alert that clears before the program can acknowledge it, and a
# second evaluation.
WAIT_SECONDS = 150

# How often the program polls the alerts, in seconds.
POLL_SECONDS = 0.5

# Returns the value of an environment variable, or a fallback if the
# variable is unset or empty, as an unset variable would be.
def env_or(name, fallback)
  value = ENV[name]
  value.nil? || value.empty? ? fallback : value
end

# Reads the settings from the environment.
def settings
  token = env_or("WORKBENCH_TOKEN", nil)
  connection_id = env_or("WORKBENCH_CONNECTION_ID", nil)
  raise "WORKBENCH_TOKEN and WORKBENCH_CONNECTION_ID must be set" unless token && connection_id

  {
    base_url: env_or("WORKBENCH_URL", "http://localhost:8080").sub(%r{/+\z}, ""),
    token: token,
    connection_id: connection_id
  }
end

# Read the settings from the environment. The :- and :? forms, unlike - and
# ?, treat an empty variable as unset.
WORKBENCH_URL="${WORKBENCH_URL:-http://localhost:8080}"
WORKBENCH_URL="${WORKBENCH_URL%/}"
: "${WORKBENCH_TOKEN:?WORKBENCH_TOKEN must be set}"
: "${WORKBENCH_CONNECTION_ID:?WORKBENCH_CONNECTION_ID must be set}"

# The rule that the script makes fire. The metric_staleness rule alerts when
# a probe's data is out of date, and the alerter checks it every minute.
RULE_NAME="metric_staleness"
# The longest the whole run waits for alerts, in seconds. The alerter
# evaluates its rules once a minute and clears an alert up to 30 seconds
# after it fires, so this allows for waiting out the alert from an earlier
# run, one alert that clears before the script can acknowledge it, and a
# second evaluation.
WAIT_SECONDS=150
# How often the script polls the alerts, in seconds.
POLL_SECONDS=0.5

def api(method: str, cfg: dict, path: str, **kwargs) -> httpx.Response:
    """Sends a request to the API and returns the response.

    A request that the server rejects raises an error that includes the
    message from the server, which raise_for_status alone would leave out.
    """
    response = httpx.request(
        method, f"{cfg['base_url']}/api/v1{path}",
        headers={"Authorization": f"Bearer {cfg['token']}"}, timeout=30,
        **kwargs,
    )
    if response.is_error:
        raise RuntimeError(
            f"{method} {path} failed with status {response.status_code}: "
            f"{response.text.strip()}"
        )
    return response

// Sends a request to the API and returns the response. A request that the
// server rejects throws an error that includes the message from the server.
async function api(
  method: string,
  cfg: Settings,
  path: string,
  body?: unknown,
): Promise<Response> {
  const headers: Record<string, string> = {
    Authorization: `Bearer ${cfg.token}`,
  };
  if (body !== undefined) headers["Content-Type"] = "application/json";

  const response = await fetch(`${cfg.baseUrl}/api/v1${path}`, {
    method,
    headers,
    body: body === undefined ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(30_000),
  });
  if (!response.ok) {
    const text = (await response.text()).trim();
    throw new Error(
      `${method} ${path} failed with status ${response.status}: ${text}`,
    );
  }
  return response;
}

/** One client serves every request; it is safe to share. */
static final HttpClient CLIENT = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(30))
        .build();

/**
 * Sends a request and returns the response with its whole body. The
 * request timeout stops applying once the headers arrive, so a server
 * that stalls part of the way through the body would hang send(); this
 * applies the same timeout to the whole exchange instead.
 */
static HttpResponse<String> send(HttpRequest request)
        throws IOException, InterruptedException {
    Duration timeout = request.timeout().orElseThrow();
    CompletableFuture<HttpResponse<String>> pending =
            CLIENT.sendAsync(request, HttpResponse.BodyHandlers.ofString());
    try {
        return pending.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        pending.cancel(true);
        throw new HttpTimeoutException("request timed out");
    } catch (ExecutionException e) {
        if (e.getCause() instanceof IOException cause) {
            throw cause;
        }
        throw new IOException(e.getCause());
    }
}

/** Converts between JSON text and Java objects. */
static final ObjectMapper JSON = new ObjectMapper();

/**
 * Sends a request to the API and returns the JSON reply. The query
 * parameters and the body may be null. A request that the server
 * rejects throws an error that includes the message from the server.
 */
static JsonNode api(String method, Settings cfg, String path,
        Map<String, String> params, Object body)
        throws IOException, InterruptedException {
    String url = cfg.baseUrl() + "/api/v1" + path;
    if (params != null) {
        url += "?" + params.entrySet().stream()
                .map(p -> URLEncoder.encode(p.getKey(),
                        StandardCharsets.UTF_8) + "="
                        + URLEncoder.encode(p.getValue(),
                                StandardCharsets.UTF_8))
                .collect(Collectors.joining("&"));
    }
    HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(url))
            .timeout(Duration.ofSeconds(30))
            .header("Authorization", "Bearer " + cfg.token());
    HttpRequest.BodyPublisher content = HttpRequest.BodyPublishers.noBody();
    if (body != null) {
        request.header("Content-Type", "application/json");
        content = HttpRequest.BodyPublishers.ofString(
                JSON.writeValueAsString(body));
    }

    HttpResponse<String> response = send(request.method(method, content).build());
    if (response.statusCode() >= 400) {
        throw new IOException(method + " " + path
                + " failed with status " + response.statusCode() + ": "
                + response.body().strip());
    }
    // Some replies have no body, which reads as a missing node.
    return JSON.readTree(response.body());
}

// One client serves every request, with a timeout for each.
var http = new HttpClient { Timeout = TimeSpan.FromSeconds(30) };
var cfg = (BaseUrl: "", Token: "", ConnectionId: "");

// Sends a request to the API and returns the JSON reply, or null if the
// reply has no body. The path may include a query string. A request that
// the server rejects throws an error that includes the message from the
// server.
async Task<JsonNode?> Api(string method, string path, object? body = null)
{
    using var request = new HttpRequestMessage(
        new HttpMethod(method), $"{cfg.BaseUrl}/api/v1{path}");
    request.Headers.Authorization =
        new AuthenticationHeaderValue("Bearer", cfg.Token);
    if (body is not null)
    {
        request.Content = JsonContent.Create(body);
    }

    using var response = await http.SendAsync(request);
    var text = await response.Content.ReadAsStringAsync();
    if ((int)response.StatusCode >= 400)
    {
        throw new HttpRequestException(
            $"{method} {path} failed with status {(int)response.StatusCode}: "
            + text.Trim());
    }
    return text.Length == 0 ? null : JsonNode.Parse(text);
}

// client gives every request a timeout; the default client has none.
var client = &http.Client{Timeout: 30 * time.Second}

// api sends a request to the API and decodes the JSON reply into out, if
// given. A request that the server rejects returns an error that includes
// the message from the server.
func api(s settings, method, path string, body, out any) error {
    var reader io.Reader
    if body != nil {
        encoded, err := json.Marshal(body)
        if err != nil {
            return err
        }
        reader = bytes.NewReader(encoded)
    }

    req, err := http.NewRequest(method, s.baseURL+"/api/v1"+path, reader)
    if err != nil {
        return err
    }
    req.Header.Set("Authorization", "Bearer "+s.token)
    if body != nil {
        req.Header.Set("Content-Type", "application/json")
    }

    resp, err := client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 400 {
        text, _ := io.ReadAll(resp.Body)
        return fmt.Errorf("%s %s failed with status %d: %s",
            method, path, resp.StatusCode, strings.TrimSpace(string(text)))
    }
    if out == nil {
        return nil
    }
    return json.NewDecoder(resp.Body).Decode(out)
}

/// One client serves every request, with a timeout for each; the client
/// has none by default.
static CLIENT: LazyLock<Client> = LazyLock::new(|| {
    Client::builder()
        .timeout(Duration::from_secs(30))
        .build()
        .expect("could not create the HTTP client")
});

/// Sends a request to the API, with any query parameters, and returns the
/// JSON reply, or null if the reply has no body. A request that the server
/// rejects returns an error that includes the message from the server.
fn api(
    method: Method,
    cfg: &Settings,
    path: &str,
    params: &[(&str, &str)],
    body: Option<&Value>,
) -> Result<Value> {
    let mut url = Url::parse(&format!("{}/api/v1{path}", cfg.base_url))?;
    // Adding no pairs would still leave an empty query string.
    if !params.is_empty() {
        url.query_pairs_mut().extend_pairs(params);
    }
    let mut request = CLIENT.request(method.clone(), url).bearer_auth(&cfg.token);
    if let Some(body) = body {
        request = request.json(body);
    }

    let response = request.send()?;
    let status = response.status().as_u16();
    let text = response.text()?;
    if status >= 400 {
        return Err(format!(
            "{method} {path} failed with status {status}: {}",
            text.trim()
        )
        .into());
    }
    if text.is_empty() {
        return Ok(Value::Null);
    }
    Ok(serde_json::from_str(&text)?)
}

# Sends a request to the API and returns the decoded reply, or nil if the
# reply has no body. The query parameters and the body are optional. A
# request that the server rejects raises an error that includes the
# message from the server.
def api(method, cfg, path, params: nil, body: nil)
  uri = URI("#{cfg[:base_url]}/api/v1#{path}")
  uri.query = URI.encode_www_form(params) if params
  headers = { "Authorization" => "Bearer #{cfg[:token]}" }
  headers["Content-Type"] = "application/json" if body

  # Net::HTTP has no overall timeout, so limit both the connection and
  # each read.
  response = Net::HTTP.start(uri.host, uri.port,
                             use_ssl: uri.scheme == "https",
                             open_timeout: 30, read_timeout: 30) do |http|
    http.send_request(method, uri.request_uri, body&.to_json, headers)
  end
  if response.code.to_i >= 400
    raise "#{method} #{path} failed with status #{response.code}: " \
          "#{response.body.to_s.strip}"
  end
  text = response.body.to_s
  text.empty? ? nil : JSON.parse(text)
end

# Sends an authenticated request and prints the JSON reply:
# api METHOD PATH [curl options]. With --fail-with-body, a request that the
# server rejects leaves the server's error message in the reply, and the
# function prints it to standard error. The token goes to curl in a header
# file that bash supplies through a pipe, so that it never appears in the
# arguments of the curl process, which other users can read in the process
# list.
api() {
    local method="$1" path="$2"
    shift 2
    local response
    response=$(curl --silent --show-error --fail-with-body \
        --connect-timeout 10 --max-time 30 \
        --request "$method" \
        --header @<(printf 'Authorization: Bearer %s\n' "$WORKBENCH_TOKEN") \
        "$@" "$WORKBENCH_URL/api/v1$path") || {
        echo "$method $path failed: $response" >&2
        return 1
    }
    printf '%s' "$response"
}

def find_rule(cfg: dict, name: str) -> int:
    """Returns the ID of an alert rule, given its name."""
    for rule in api("GET", cfg, "/alert-rules").json():
        if rule["name"] == name:
            return rule["id"]
    raise RuntimeError(f"There is no alert rule named '{name}'")

interface Rule {
  id: number;
  name: string;
}

// Returns the ID of an alert rule, given its name.
async function findRule(cfg: Settings, name: string): Promise<number> {
  const rules = (await (await api("GET", cfg, "/alert-rules")).json()) as Rule[];
  const rule = rules.find((r) => r.name === name);
  if (!rule) throw new Error(`There is no alert rule named '${name}'`);
  return rule.id;
}

/** Returns the ID of an alert rule, given its name. */
static long findRule(Settings cfg, String name)
        throws IOException, InterruptedException {
    for (JsonNode rule : api("GET", cfg, "/alert-rules", null, null)) {
        if (name.equals(rule.path("name").asString())) {
            return rule.get("id").asLong();
        }
    }
    throw new IOException("There is no alert rule named '" + name + "'");
}

// Returns the ID of an alert rule, given its name.
async Task<long> FindRule(string name)
{
    var rules = (await Api("GET", "/alert-rules"))?.AsArray()
        ?? throw new InvalidOperationException("The reply had no rules");
    var rule = rules.FirstOrDefault(
        r => r?["name"]?.GetValue<string>() == name)
        ?? throw new InvalidOperationException(
            $"There is no alert rule named '{name}'");
    return rule["id"]!.GetValue<long>();
}

// findRule returns the ID of an alert rule, given its name.
func findRule(s settings, name string) (int64, error) {
    var rules []struct {
        ID   int64  `json:"id"`
        Name string `json:"name"`
    }
    if err := api(s, "GET", "/alert-rules", nil, &rules); err != nil {
        return 0, err
    }
    for _, rule := range rules {
        if rule.Name == name {
            return rule.ID, nil
        }
    }
    return 0, fmt.Errorf("there is no alert rule named '%s'", name)
}

/// Returns the ID of an alert rule, given its name.
fn find_rule(cfg: &Settings, name: &str) -> Result<i64> {
    let rules = api(Method::GET, cfg, "/alert-rules", &[], None)?;
    rules
        .as_array()
        .and_then(|rules| rules.iter().find(|r| r["name"] == name))
        .and_then(|rule| rule["id"].as_i64())
        .ok_or_else(|| format!("there is no alert rule named '{name}'").into())
}

# Returns the ID of an alert rule, given its name.
def find_rule(cfg, name)
  rule = api("GET", cfg, "/alert-rules").find { |r| r["name"] == name }
  raise "There is no alert rule named '#{name}'" if rule.nil?

  rule.fetch("id")
end

# Prints the ID of an alert rule, given its name. The function fails when
# there is no such rule, because jq prints nothing for it.
find_rule() {
    local name="$1" id
    id=$(api GET /alert-rules |
        jq -r --arg name "$name" 'first(.[] | select(.name == $name)) | .id')
    if [[ -z "$id" ]]; then
        echo "There is no alert rule named '$name'" >&2
        return 1
    fi
    printf '%s' "$id"
}

def set_override(cfg: dict, rule_id: int) -> None:
    """Overrides the threshold of a rule for one server.

    A threshold of -1 is below any value that the staleness ratio can
    take, so the rule is violated on the next evaluation.
    """
    api("PUT", cfg,
        f"/alert-overrides/server/{cfg['connection_id']}/{rule_id}",
        json={"operator": ">", "threshold": -1, "severity": "info",
              "enabled": True})

// Overrides the threshold of a rule for one server. A threshold of -1 is
// below any value that the staleness ratio can take, so the rule is violated
// on the next evaluation.
async function setOverride(cfg: Settings, ruleId: number): Promise<void> {
  await api(
    "PUT",
    cfg,
    `/alert-overrides/server/${cfg.connectionId}/${ruleId}`,
    { operator: ">", threshold: -1, severity: "info", enabled: true },
  );
}

/**
 * Overrides the threshold of a rule for one server. A threshold of -1
 * is below any value that the staleness ratio can take, so the rule is
 * violated on the next evaluation.
 */
static void setOverride(Settings cfg, long ruleId)
        throws IOException, InterruptedException {
    api("PUT", cfg, "/alert-overrides/server/" + cfg.connectionId()
            + "/" + ruleId, null, Map.of(
                    "operator", ">",
                    "threshold", -1,
                    "severity", "info",
                    "enabled", true));
}

// Overrides the threshold of a rule for one server. A threshold of -1 is
// below any value that the staleness ratio can take, so the rule is
// violated on the next evaluation.
async Task SetOverride(long ruleId) =>
    await Api("PUT", $"/alert-overrides/server/{cfg.ConnectionId}/{ruleId}",
        new { @operator = ">", threshold = -1, severity = "info",
            enabled = true });

// setOverride overrides the threshold of a rule for one server. A threshold
// of -1 is below any value that the staleness ratio can take, so the rule is
// violated on the next evaluation.
func setOverride(s settings, ruleID int64) error {
    path := fmt.Sprintf("/alert-overrides/server/%s/%d", s.connectionID, ruleID)
    return api(s, "PUT", path, map[string]any{
        "operator": ">", "threshold": -1, "severity": "info", "enabled": true,
    }, nil)
}

/// Overrides the threshold of a rule for one server. A threshold of -1 is
/// below any value that the staleness ratio can take, so the rule is
/// violated on the next evaluation.
fn set_override(cfg: &Settings, rule_id: i64) -> Result<()> {
    let path = format!("/alert-overrides/server/{}/{rule_id}", cfg.connection_id);
    let body = json!({ "operator": ">", "threshold": -1, "severity": "info", "enabled": true });
    api(Method::PUT, cfg, &path, &[], Some(&body))?;
    Ok(())
}

# Overrides the threshold of a rule for one server. A threshold of -1 is
# below any value that the staleness ratio can take, so the rule is
# violated on the next evaluation.
def set_override(cfg, rule_id)
  api("PUT", cfg, "/alert-overrides/server/#{cfg[:connection_id]}/#{rule_id}",
      body: { operator: ">", threshold: -1, severity: "info", enabled: true })
end

# Overrides the threshold of a rule for one server. A threshold of -1 is
# below any value that the staleness ratio can take, so the rule is
# violated on the next evaluation.
set_override() {
    local rule_id="$1"
    api PUT "/alert-overrides/server/$WORKBENCH_CONNECTION_ID/$rule_id" \
        --header 'Content-Type: application/json' \
        --data '{"operator": ">", "threshold": -1, "severity": "info", "enabled": true}' \
        >/dev/null
}

def active_alerts(cfg: dict, rule_id: int) -> list:
    """Returns the active alerts that the given rule raised for the server.

    The request filters by server, status, and type, and asks for the most
    alerts that the API returns at once, so that the result does not depend
    on the first 100 alerts of a busy Workbench.
    """
    alerts = api("GET", cfg, "/alerts", params={
        "connection_id": cfg["connection_id"], "status": "active",
        "alert_type": "threshold", "limit": 1000,
    }).json()["alerts"]
    return [alert for alert in alerts if alert["rule_id"] == rule_id]


def wait_until_quiet(cfg: dict, rule_id: int, deadline: float) -> None:
    """Waits until the rule has no active alert.

    The alerter raises no new alert for a rule while one is active, so an
    alert left by an earlier run, which clears within 30 seconds, would
    otherwise stand in for the one that this run causes.
    """
    while active_alerts(cfg, rule_id):
        if time.monotonic() >= deadline:
            raise RuntimeError("An earlier alert did not clear in time")
        time.sleep(POLL_SECONDS)


def wait_for_alert(cfg: dict, rule_id: int, deadline: float) -> dict:
    """Polls until the given rule raises an alert, and returns it.

    The alert clears itself within 30 seconds of firing, so the program
    polls twice a second.
    """
    while True:
        alerts = active_alerts(cfg, rule_id)
        if alerts:
            return alerts[0]
        if time.monotonic() >= deadline:
            raise RuntimeError(
                f"No alert was raised within {WAIT_SECONDS} seconds"
            )
        time.sleep(POLL_SECONDS)

interface Alert {
  id: number;
  rule_id: number;
  severity: string;
  status: string;
}

// Returns the threshold alerts for a server that have the given status. The
// request filters by server, status, and type, and asks for the most alerts
// that the API returns at once, so that the result does not depend on the
// first 100 alerts of a busy Workbench.
async function listAlerts(cfg: Settings, status: string): Promise<Alert[]> {
  const query = new URLSearchParams({
    connection_id: cfg.connectionId,
    status,
    alert_type: "threshold",
    limit: "1000",
  });
  const response = await api("GET", cfg, `/alerts?${query}`);
  return ((await response.json()) as { alerts: Alert[] }).alerts;
}

// Returns the active alerts that the given rule raised.
async function activeAlerts(cfg: Settings, ruleId: number): Promise<Alert[]> {
  return (await listAlerts(cfg, "active")).filter((a) => a.rule_id === ruleId);
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// Waits until the rule has no active alert. The alerter raises no new alert
// for a rule while one is active, so an alert left by an earlier run, which
// clears within 30 seconds, would otherwise stand in for the one that this
// run causes.
async function waitUntilQuiet(
  cfg: Settings,
  ruleId: number,
  deadline: number,
): Promise<void> {
  while ((await activeAlerts(cfg, ruleId)).length > 0) {
    if (Date.now() >= deadline) {
      throw new Error("An earlier alert did not clear in time");
    }
    await sleep(POLL_MS);
  }
}

// Polls until the given rule raises an alert, and returns it. The alert
// clears itself within 30 seconds of firing, so the program polls twice a
// second.
async function waitForAlert(
  cfg: Settings,
  ruleId: number,
  deadline: number,
): Promise<Alert> {
  for (;;) {
    const [alert] = await activeAlerts(cfg, ruleId);
    if (alert) return alert;
    if (Date.now() >= deadline) {
      throw new Error(`No alert was raised within ${WAIT_MS / 1000} seconds`);
    }
    await sleep(POLL_MS);
  }
}

/**
 * Returns the threshold alerts for the server that have the given
 * status. The request asks for the most alerts that the API returns at
 * once, so that the result does not depend on the first 100 alerts of
 * a busy Workbench.
 */
static List<JsonNode> listAlerts(Settings cfg, String status)
        throws IOException, InterruptedException {
    Map<String, String> params = new LinkedHashMap<>();
    params.put("connection_id", cfg.connectionId());
    params.put("status", status);
    params.put("alert_type", "threshold");
    params.put("limit", "1000");
    List<JsonNode> alerts = new ArrayList<>();
    api("GET", cfg, "/alerts", params, null).path("alerts")
            .forEach(alerts::add);
    return alerts;
}

/** Returns the active alerts that the given rule raised for the server. */
static List<JsonNode> activeAlerts(Settings cfg, long ruleId)
        throws IOException, InterruptedException {
    List<JsonNode> alerts = listAlerts(cfg, "active");
    alerts.removeIf(alert -> alert.path("rule_id").asLong() != ruleId);
    return alerts;
}

/**
 * Waits until the rule has no active alert. The alerter raises no new
 * alert for a rule while one is active, so an alert left by an earlier
 * run, which clears within 30 seconds, would otherwise stand in for the
 * one that this run causes.
 */
static void waitUntilQuiet(Settings cfg, long ruleId, long deadline)
        throws IOException, InterruptedException {
    while (!activeAlerts(cfg, ruleId).isEmpty()) {
        if (System.nanoTime() - deadline >= 0) {
            throw new IOException("An earlier alert did not clear in time");
        }
        Thread.sleep(POLL_INTERVAL.toMillis());
    }
}

/**
 * Polls until the given rule raises an alert, and returns it. The alert
 * clears itself within 30 seconds of firing, so the program polls twice
 * a second.
 */
static JsonNode waitForAlert(Settings cfg, long ruleId, long deadline)
        throws IOException, InterruptedException {
    while (true) {
        List<JsonNode> alerts = activeAlerts(cfg, ruleId);
        if (!alerts.isEmpty()) {
            return alerts.get(0);
        }
        if (System.nanoTime() - deadline >= 0) {
            throw new IOException("No alert was raised within "
                    + WAIT_TIME.toSeconds() + " seconds");
        }
        Thread.sleep(POLL_INTERVAL.toMillis());
    }
}

// Returns the threshold alerts for the server that have the given status.
// The request asks for the most alerts that the API returns at once, so
// that the result does not depend on the first 100 alerts of a busy
// Workbench.
async Task<List<JsonNode>> ListAlerts(string status)
{
    var query = $"connection_id={Uri.EscapeDataString(cfg.ConnectionId)}"
        + $"&status={status}&alert_type=threshold&limit=1000";
    var reply = await Api("GET", $"/alerts?{query}");
    return reply?["alerts"]?.AsArray().OfType<JsonNode>().ToList()
        ?? throw new InvalidOperationException("The reply had no alerts");
}

// Returns the active alerts that the given rule raised for the server.
async Task<List<JsonNode>> ActiveAlerts(long ruleId) =>
    (await ListAlerts("active"))
        .Where(a => a["rule_id"]?.GetValue<long>() == ruleId)
        .ToList();

// Waits until the rule has no active alert. The alerter raises no new
// alert for a rule while one is active, so an alert left by an earlier
// run, which clears within 30 seconds, would otherwise stand in for the
// one that this run causes.
async Task WaitUntilQuiet(long ruleId, Stopwatch clock)
{
    while ((await ActiveAlerts(ruleId)).Count > 0)
    {
        if (clock.Elapsed >= waitTime)
        {
            throw new InvalidOperationException(
                "An earlier alert did not clear in time");
        }
        await Task.Delay(pollInterval);
    }
}

// Polls until the given rule raises an alert, and returns it. The alert
// clears itself within 30 seconds of firing, so the program polls twice a
// second.
async Task<JsonNode> WaitForAlert(long ruleId, Stopwatch clock)
{
    while (true)
    {
        var alerts = await ActiveAlerts(ruleId);
        if (alerts.Count > 0)
        {
            return alerts[0];
        }
        if (clock.Elapsed >= waitTime)
        {
            throw new InvalidOperationException(
                $"No alert was raised within {waitTime.TotalSeconds} seconds");
        }
        await Task.Delay(pollInterval);
    }
}

// alert is the part of an alert that the program needs.
type alert struct {
    ID       int64  `json:"id"`
    RuleID   int64  `json:"rule_id"`
    Severity string `json:"severity"`
    Status   string `json:"status"`
}

// listAlerts returns the threshold alerts for a server that have the given
// status. The request filters by server, status, and type, and asks for the
// most alerts that the API returns at once, so that the result does not
// depend on the first 100 alerts of a busy Workbench.
func listAlerts(s settings, status string) ([]alert, error) {
    query := url.Values{
        "connection_id": {s.connectionID},
        "status":        {status},
        "alert_type":    {"threshold"},
        "limit":         {"1000"},
    }
    var result struct {
        Alerts []alert `json:"alerts"`
    }
    err := api(s, "GET", "/alerts?"+query.Encode(), nil, &result)
    return result.Alerts, err
}

// activeAlerts returns the active alerts that the given rule raised.
func activeAlerts(s settings, ruleID int64) ([]alert, error) {
    alerts, err := listAlerts(s, "active")
    var raised []alert
    for _, a := range alerts {
        if a.RuleID == ruleID {
            raised = append(raised, a)
        }
    }
    return raised, err
}

// waitUntilQuiet waits until the rule has no active alert. The alerter
// raises no new alert for a rule while one is active, so an alert left by an
// earlier run, which clears within 30 seconds, would otherwise stand in for
// the one that this run causes.
func waitUntilQuiet(s settings, ruleID int64, deadline time.Time) error {
    for {
        raised, err := activeAlerts(s, ruleID)
        if err != nil {
            return err
        }
        if len(raised) == 0 {
            return nil
        }
        if time.Now().After(deadline) {
            return errors.New("an earlier alert did not clear in time")
        }
        time.Sleep(pollTime)
    }
}

// waitForAlert polls until the given rule raises an alert, and returns it.
// The alert clears itself within 30 seconds of firing, so the program polls
// twice a second.
func waitForAlert(s settings, ruleID int64, deadline time.Time) (alert, error) {
    for {
        raised, err := activeAlerts(s, ruleID)
        if err != nil {
            return alert{}, err
        }
        if len(raised) > 0 {
            return raised[0], nil
        }
        if time.Now().After(deadline) {
            return alert{}, fmt.Errorf("no alert was raised within %s", waitTime)
        }
        time.Sleep(pollTime)
    }
}

/// Returns the threshold alerts for the server that have the given status.
/// The request asks for the most alerts that the API returns at once, so
/// that the result does not depend on the first 100 alerts of a busy
/// Workbench.
fn list_alerts(cfg: &Settings, status: &str) -> Result<Vec<Value>> {
    let params = [
        ("connection_id", cfg.connection_id.as_str()),
        ("status", status),
        ("alert_type", "threshold"),
        ("limit", "1000"),
    ];
    let mut reply = api(Method::GET, cfg, "/alerts", &params, None)?;
    match reply["alerts"].take() {
        Value::Array(alerts) => Ok(alerts),
        _ => Err("the reply had no alerts".into()),
    }
}

/// Returns the active alerts that the given rule raised for the server.
fn active_alerts(cfg: &Settings, rule_id: i64) -> Result<Vec<Value>> {
    let mut alerts = list_alerts(cfg, "active")?;
    alerts.retain(|alert| alert["rule_id"] == rule_id);
    Ok(alerts)
}

/// Waits until the rule has no active alert. The alerter raises no new
/// alert for a rule while one is active, so an alert left by an earlier
/// run, which clears within 30 seconds, would otherwise stand in for the
/// one that this run causes.
fn wait_until_quiet(cfg: &Settings, rule_id: i64, deadline: Instant) -> Result<()> {
    while !active_alerts(cfg, rule_id)?.is_empty() {
        if Instant::now() >= deadline {
            return Err("an earlier alert did not clear in time".into());
        }
        thread::sleep(POLL_INTERVAL);
    }
    Ok(())
}

/// Polls until the given rule raises an alert, and returns it. The alert
/// clears itself within 30 seconds of firing, so the program polls twice a
/// second.
fn wait_for_alert(cfg: &Settings, rule_id: i64, deadline: Instant) -> Result<Value> {
    loop {
        if let Some(alert) = active_alerts(cfg, rule_id)?.into_iter().next() {
            return Ok(alert);
        }
        if Instant::now() >= deadline {
            let seconds = WAIT_TIME.as_secs();
            return Err(format!("no alert was raised within {seconds} seconds").into());
        }
        thread::sleep(POLL_INTERVAL);
    }
}