Changing a Database's Settings

# httpx, an HTTP client with a requests-like API.
httpx>=0.28,<1
# psycopg 3, with the pre-built binary package that bundles libpq.
psycopg[binary]>=3.2,<4

{
  "name": "update-database",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/main.js"
  },
  "dependencies": {
    "pg": "^8.23.1"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "@types/pg": "^8.23.1",
    "typescript": "^5.9.0"
  }
}

<dependencies>
  <!-- Jackson, for reading and writing 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>
  <!-- The PostgreSQL JDBC driver (pgJDBC). -->
  <dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.7.14</version>
  </dependency>
</dependencies>

<ItemGroup>
  <!-- Npgsql, the ADO.NET data provider for PostgreSQL. HttpClient and
       System.Text.Json are part of .NET. -->
  <PackageReference Include="Npgsql" Version="10.0.3" />
</ItemGroup>

// pgx is a pure Go driver and toolkit for PostgreSQL.
require github.com/jackc/pgx/v5 v5.12.0

require (
    github.com/jackc/pgpassfile v1.0.0 // indirect
    github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
    golang.org/x/text v0.41.0 // indirect
)

[dependencies]
# The blocking client, with TLS from rustls and request and response bodies
# as JSON; the default features add HTTP/2 and other things this example
# does not use.
reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] }
serde_json = "1"
# The synchronous PostgreSQL client; tokio-postgres is the asynchronous
# equivalent.
postgres = "0.19"
# Handling Ctrl-C, and with the termination feature SIGTERM, so that the
# program can restore the settings before it exits.
ctrlc = { version = "3", features = ["termination"] }

source "https://rubygems.org"

# Net::HTTP and json are part of the standard library. The pg gem wraps
# libpq, so libpq and its headers must be installed.
gem "pg", "~> 1.6"

import copy
import os
import signal
import sys
import time

import httpx
import psycopg

// Node.js provides fetch and AbortSignal as globals; setTimeout from
// timers/promises returns a promise, which suits a polling loop. pg is
// node-postgres.
import { setTimeout as sleep } from "node:timers/promises";
import pg from "pg";

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpTimeoutException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import tools.jackson.core.JacksonException;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.json.JsonMapper;
import tools.jackson.databind.node.ObjectNode;

using System.Diagnostics;
using System.Net.Http.Json;
using System.Runtime.InteropServices;
using System.Text.Json;
using System.Text.Json.Nodes;
using Npgsql;

import (
    "bytes"
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "maps"
    "net"
    "net/http"
    "net/url"
    "os"
    "os/signal"
    "strconv"
    "strings"
    "syscall"
    "time"

    "github.com/jackc/pgx/v5"
)

use std::env;
use std::error::Error;
use std::fmt;
use std::process::ExitCode;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use postgres::{Config, NoTls};
use reqwest::blocking::Client;
use reqwest::Method;
use serde_json::{json, Value};

require "json"
require "net/http"
require "pg"
require "uri"

# Stop at the first failed command or unset variable, and make a failure
# anywhere in a pipeline fail the whole pipeline.
set -euo pipefail

# The script needs curl to call the API, jq to read and build the JSON,
# and psql to connect to the database.
for tool in curl jq psql; do
    command -v "$tool" >/dev/null || { echo "Control Plane error: $tool is required" >&2; exit 1; }
done

# The or operator treats an empty variable as unset. The password has no
# default; the Control Plane never returns it, so it comes from whoever
# created the database user.
BASE_URL = (os.environ.get("CONTROL_PLANE_URL") or "http://localhost:3000").rstrip("/")
DATABASE_ID = os.environ.get("CONTROL_PLANE_DATABASE_ID") or "cookbook"
DATABASE_USER = os.environ.get("CONTROL_PLANE_DATABASE_USER") or "cookbook"
DATABASE_PASSWORD = os.environ.get("CONTROL_PLANE_DATABASE_PASSWORD") or None

TIMEOUT_SECONDS = 30

# The setting to change, and its new value.
SETTING = "work_mem"
VALUE = "16MB"

# How long to wait for an update to finish, and how often to check.
WAIT_SECONDS = 600
POLL_SECONDS = 5


class ControlPlaneError(Exception):
    """Raised when the Control Plane rejects a request or replies oddly.
    The status is the HTTP status of the reply, if there was one."""

    def __init__(self, message: str, status: int | None = None):
        super().__init__(message)
        self.status = status


def call(client: httpx.Client, method: str, path: str, body: dict | None = None) -> dict:
    """Sends a request to the server, and returns the decoded reply."""
    response = client.request(method, path, json=body)
    if response.is_error:
        raise ControlPlaneError(
            f"HTTP {response.status_code}, {response.text[:200]}",
            response.status_code)
    try:
        reply = response.json()
        if isinstance(reply, dict):
            return reply
    except ValueError:
        pass
    raise ControlPlaneError(f"Unexpected response: {response.text[:200]}")

// The || operator, unlike ??, also treats an empty variable as unset. The
// password has no default; the Control Plane never returns it, so it
// comes from whoever created the database user.
const baseUrl = (process.env.CONTROL_PLANE_URL || "http://localhost:3000").replace(/\/+$/, "");
const databaseId = process.env.CONTROL_PLANE_DATABASE_ID || "cookbook";
const databaseUser = process.env.CONTROL_PLANE_DATABASE_USER || "cookbook";
const databasePassword = process.env.CONTROL_PLANE_DATABASE_PASSWORD || undefined;

const timeoutMs = 30_000;

// The setting to change, and its new value.
const setting = "work_mem";
const value = "16MB";

// How long to wait for an update to finish, and how often to check.
const waitMs = 600_000;
const pollMs = 5_000;

// Ctrl-C and SIGTERM abort the signal below instead of ending the process,
// which lets the program restore the settings before it exits. The
// handlers stay in place, so a further Ctrl-C does not end the process
// during the cleanup either.
const interrupt = new AbortController();
for (const signalName of ["SIGINT", "SIGTERM"] as const) {
  process.on(signalName, () => interrupt.abort());
}

// An error reply from the Control Plane, with its HTTP status.
class ControlPlaneError extends Error {
  constructor(
    message: string,
    readonly status?: number,
  ) {
    super(message);
  }
}

// An error from the database connection or its query.
class DatabaseError extends Error {}

// Sends a request to the server, and returns the decoded reply. The
// request stops when the user presses Ctrl-C, unless the caller passes a
// different signal, as the cleanup does.
async function call<T>(
  method: string,
  path: string,
  body?: unknown,
  signal: AbortSignal = interrupt.signal,
): Promise<T> {
  const response = await fetch(`${baseUrl}${path}`, {
    method,
    headers: body === undefined ? {} : { "Content-Type": "application/json" },
    body: body === undefined ? undefined : JSON.stringify(body),
    signal: AbortSignal.any([signal, AbortSignal.timeout(timeoutMs)]),
  });

  // fetch only rejects when it cannot reach the server; an error status
  // is a normal response, so check it explicitly.
  const text = await response.text();
  if (!response.ok) {
    throw new ControlPlaneError(
      `HTTP ${response.status}, ${text.slice(0, 200) || response.statusText}`,
      response.status,
    );
  }
  try {
    return JSON.parse(text) as T;
  } catch {
    throw new ControlPlaneError(`Unexpected response: ${text.slice(0, 200)}`);
  }
}

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

// The password has no default; the Control Plane never returns it, so
// it comes from whoever created the database user.
static final String BASE_URL =
        envOr("CONTROL_PLANE_URL", "http://localhost:3000")
                .replaceAll("/+$", "");
static final String DATABASE_ID = envOr("CONTROL_PLANE_DATABASE_ID", "cookbook");
static final String DATABASE_USER = envOr("CONTROL_PLANE_DATABASE_USER", "cookbook");
static final String DATABASE_PASSWORD = envOr("CONTROL_PLANE_DATABASE_PASSWORD", null);

static final Duration TIMEOUT = Duration.ofSeconds(30);

// The setting to change, and its new value.
static final String SETTING = "work_mem";
static final String VALUE = "16MB";

// How long to wait for an update to finish, and how often to check.
static final Duration WAIT_FOR = Duration.ofSeconds(600);
static final Duration POLL_INTERVAL = Duration.ofSeconds(5);

/**
 * Thrown when the Control Plane rejects a request or replies oddly.
 * The status is the HTTP status of the reply, or 0 if there was none.
 */
static class ControlPlaneException extends Exception {
    final int status;

    ControlPlaneException(String message, int status) {
        super(message);
        this.status = status;
    }

    ControlPlaneException(String message) {
        this(message, 0);
    }
}

// A client and a mapper are safe to share, so the program creates only
// one of each.
static final HttpClient CLIENT = HttpClient.newHttpClient();
static final ObjectMapper MAPPER = JsonMapper.builder().build();

/** Returns at most the first 200 characters of a reply. */
static String shorten(String text) {
    return text.length() > 200 ? text.substring(0, 200) : text;
}

/**
 * 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 {
    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());
    }
}

/**
 * Sends a request to the server, with the body encoded as JSON if it
 * is not null, and returns the decoded reply.
 */
static JsonNode call(String method, String path, Object body)
        throws IOException, InterruptedException, ControlPlaneException {
    HttpRequest.Builder builder = HttpRequest.newBuilder(URI.create(BASE_URL + path))
            .timeout(TIMEOUT);
    if (body == null) {
        builder.method(method, HttpRequest.BodyPublishers.noBody());
    } else {
        builder.header("Content-Type", "application/json")
                .method(method, HttpRequest.BodyPublishers.ofString(
                        MAPPER.writeValueAsString(body)));
    }
    HttpResponse<String> response = send(builder.build());
    if (response.statusCode() >= 400) {
        throw new ControlPlaneException("HTTP " + response.statusCode()
                + ", " + shorten(response.body()), response.statusCode());
    }
    try {
        JsonNode reply = MAPPER.readTree(response.body());
        if (reply.isObject()) {
            return reply;
        }
    } catch (JacksonException e) {
        // The body is not JSON; report it below.
    }
    throw new ControlPlaneException(
            "Unexpected response: " + shorten(response.body()));
}

// Returns the value of an environment variable, or a fallback if the
// variable is unset or empty.
static string? EnvOr(string name, string? fallback) =>
    Environment.GetEnvironmentVariable(name) is { Length: > 0 } value
        ? value
        : fallback;

static string BaseUrl() =>
    EnvOr("CONTROL_PLANE_URL", "http://localhost:3000")!.TrimEnd('/');

static string DatabaseId() => EnvOr("CONTROL_PLANE_DATABASE_ID", "cookbook")!;

// The password has no default; the Control Plane never returns it, so it
// comes from whoever created the database user.
static string DatabaseUser() => EnvOr("CONTROL_PLANE_DATABASE_USER", "cookbook")!;
static string? DatabasePassword() => EnvOr("CONTROL_PLANE_DATABASE_PASSWORD", null);

const int TimeoutSeconds = 30;

// The setting to change, and its new value.
const string Setting = "work_mem";
const string Value = "16MB";

// How long to wait for an update to finish, and how often to check.
const int WaitSeconds = 600;
const int PollSeconds = 5;

// Returns at most the first 200 characters of a reply.
static string Shorten(string text) =>
    text.Length > 200 ? text[..200] : text;

// Sends a request to the server, with the body encoded as JSON if it is
// not null, and returns the decoded reply. The token stops the request
// when the user presses Ctrl-C.
static async Task<JsonObject> CallAsync(
    HttpClient client, HttpMethod method, string path, object? body,
    CancellationToken token)
{
    using var request = new HttpRequestMessage(method, BaseUrl() + path);
    if (body is not null)
    {
        request.Content = JsonContent.Create(body);
    }
    using var response = await client.SendAsync(request, token);
    var text = await response.Content.ReadAsStringAsync(token);
    if (!response.IsSuccessStatusCode)
    {
        throw new ControlPlaneException(
            $"HTTP {(int)response.StatusCode}, {Shorten(text)}",
            (int)response.StatusCode);
    }
    try
    {
        if (JsonNode.Parse(text) is JsonObject reply)
        {
            return reply;
        }
    }
    catch (JsonException)
    {
        // The body is not JSON; report it below.
    }
    throw new ControlPlaneException($"Unexpected response: {Shorten(text)}");
}

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

// The password has no default; the Control Plane never returns it, so it
// comes from whoever created the database user.
var (
    baseURL          = strings.TrimRight(envOr("CONTROL_PLANE_URL", "http://localhost:3000"), "/")
    databaseID       = envOr("CONTROL_PLANE_DATABASE_ID", "cookbook")
    databaseUser     = envOr("CONTROL_PLANE_DATABASE_USER", "cookbook")
    databasePassword = os.Getenv("CONTROL_PLANE_DATABASE_PASSWORD")
)

const timeout = 30 * time.Second

// The setting to change, and its new value.
const (
    setting = "work_mem"
    value   = "16MB"
)

// How long to wait for an update to finish, and how often to check.
const (
    waitFor      = 600 * time.Second
    pollInterval = 5 * time.Second
)

// apiError is an error reply from the Control Plane, with its HTTP status.
type apiError struct {
    status int
    body   string
}

func (e *apiError) Error() string {
    return fmt.Sprintf("HTTP %d, %.200s", e.status, e.body)
}

// statusOf returns the HTTP status of an error reply, or 0 if no reply
// arrived.
func statusOf(err error) int {
    var apiErr *apiError
    if errors.As(err, &apiErr) {
        return apiErr.status
    }
    return 0
}

// dbError marks an error from the database connection or its query.
type dbError struct{ err error }

func (e *dbError) Error() string { return e.err.Error() }

var client = &http.Client{Timeout: timeout}

// call sends a request to the server, and decodes the reply into out if
// out is not nil. The context stops the request when the user presses
// Ctrl-C.
func call(ctx context.Context, 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)
    }
    request, err := http.NewRequestWithContext(ctx, method, baseURL+path, reader)
    if err != nil {
        return err
    }
    if body != nil {
        request.Header.Set("Content-Type", "application/json")
    }

    response, err := client.Do(request)
    if err != nil {
        return err
    }
    defer response.Body.Close()
    reply, err := io.ReadAll(response.Body)
    if err != nil {
        return err
    }

    // Go treats an error status as a normal response, so check it here.
    if response.StatusCode >= 400 {
        return &apiError{response.StatusCode, strings.TrimSpace(string(reply))}
    }
    if out != nil {
        if err := json.Unmarshal(reply, out); err != nil {
            return fmt.Errorf("unexpected response: %.200s", reply)
        }
    }
    return nil
}

// sleep waits for the poll interval, or until the context ends.
func sleep(ctx context.Context) error {
    select {
    case <-time.After(pollInterval):
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

/// Returns the value of an environment variable, or a fallback if the
/// variable is unset or empty.
fn env_or(name: &str, fallback: &str) -> String {
    env::var(name)
        .ok()
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| fallback.to_string())
}

fn base_url() -> String {
    env_or("CONTROL_PLANE_URL", "http://localhost:3000")
        .trim_end_matches('/')
        .to_string()
}

fn database_id() -> String {
    env_or("CONTROL_PLANE_DATABASE_ID", "cookbook")
}

const TIMEOUT: Duration = Duration::from_secs(30);

// The setting to change, and its new value.
const SETTING: &str = "work_mem";
const VALUE: &str = "16MB";

// How long to wait for an update to finish, and how often to check.
const WAIT_FOR: Duration = Duration::from_secs(600);
const POLL_INTERVAL: Duration = Duration::from_secs(5);

/// Set by Ctrl-C or SIGTERM. The handler sets this flag instead of ending
/// the process, which lets the program restore the settings before it
/// exits; the waits check it between requests.
static INTERRUPTED: AtomicBool = AtomicBool::new(false);

/// The errors that the functions below return; a string converts into one
/// with the ? operator or into().
type BoxError = Box<dyn Error>;

/// An error reply from the Control Plane, with its HTTP status.
#[derive(Debug)]
struct ApiError {
    status: u16,
    body: String,
}

impl fmt::Display for ApiError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "HTTP {}, {}", self.status, self.body)
    }
}

impl Error for ApiError {}

/// Returns the HTTP status of an error reply, or None if no reply arrived
/// or the error is of another kind.
fn status_of(error: &BoxError) -> Option<u16> {
    error.downcast_ref::<ApiError>().map(|e| e.status)
}

/// Returns at most the first 200 characters of a reply.
fn shorten(text: &str) -> String {
    text.chars().take(200).collect()
}

/// Sends a request to the server, with the body encoded as JSON if there
/// is one, and returns the decoded reply.
fn call(client: &Client, method: Method, path: &str, body: Option<&Value>) -> Result<Value, BoxError> {
    let mut request = client.request(method, format!("{}{path}", base_url()));
    if let Some(body) = body {
        request = request.json(body);
    }
    let response = request.send()?;
    let status = response.status();
    let text = response.text()?;
    if !status.is_success() {
        return Err(ApiError { status: status.as_u16(), body: shorten(&text) }.into());
    }
    match serde_json::from_str::<Value>(&text) {
        Ok(reply) if reply.is_object() => Ok(reply),
        _ => Err(format!("Unexpected response: {}", shorten(&text)).into()),
    }
}

/// Fails if check is true and the user has pressed Ctrl-C. A request does
/// not stop for Ctrl-C, so the program checks the flag after each one; the
/// cleanup passes false, so that it carries on.
fn check_interrupted(check: bool) -> Result<(), BoxError> {
    if check && INTERRUPTED.load(Ordering::SeqCst) {
        return Err("interrupted".into());
    }
    Ok(())
}

/// Waits for the poll interval, checking for Ctrl-C as above.
fn pause(check: bool) -> Result<(), BoxError> {
    let end = Instant::now() + POLL_INTERVAL;
    while Instant::now() < end {
        check_interrupted(check)?;
        thread::sleep(Duration::from_millis(100));
    }
    Ok(())
}

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

# The password has no default; the Control Plane never returns it, so it
# comes from whoever created the database user.
BASE_URL = env_or("CONTROL_PLANE_URL", "http://localhost:3000").sub(%r{/+\z}, "")
DATABASE_ID = env_or("CONTROL_PLANE_DATABASE_ID", "cookbook")
DATABASE_USER = env_or("CONTROL_PLANE_DATABASE_USER", "cookbook")
DATABASE_PASSWORD = env_or("CONTROL_PLANE_DATABASE_PASSWORD", nil)

TIMEOUT_SECONDS = 30

# The setting to change, and its new value.
SETTING = "work_mem"
VALUE = "16MB"

# How long to wait for an update to finish, and how often to check.
WAIT_SECONDS = 600
POLL_SECONDS = 5

# Raised when the Control Plane rejects a request or replies oddly. The
# status is the HTTP status of the reply, if there was one.
class ControlPlaneError < StandardError
  attr_reader :status

  def initialize(message, status = nil)
    super(message)
    @status = status
  end
end

# The errors that mean no reply arrived: SystemCallError covers a refused
# connection, Timeout::Error covers the connection and read timeouts, and
# SocketError covers an unknown host.
NETWORK_ERRORS = [SystemCallError, Timeout::Error, SocketError, IOError].freeze

# Sends a request to the server, and returns the decoded reply.
def call(method, path, body = nil)
  uri = URI("#{BASE_URL}#{path}")
  request = Net::HTTPGenericRequest.new(method, !body.nil?, true, uri)
  if body
    request["Content-Type"] = "application/json"
    request.body = JSON.generate(body)
  end
  # start opens a connection, and closes it when the block exits, whether
  # or not an error occurred.
  response = Net::HTTP.start(
    uri.host, uri.port,
    use_ssl: uri.scheme == "https",
    open_timeout: TIMEOUT_SECONDS, read_timeout: TIMEOUT_SECONDS
  ) { |http| http.request(request) }
  unless response.is_a?(Net::HTTPSuccess)
    raise ControlPlaneError.new(
      "HTTP #{response.code}, #{response.body.to_s[0, 200]}", response.code.to_i
    )
  end

  begin
    reply = JSON.parse(response.body.to_s)
    return reply if reply.is_a?(Hash)
  rescue JSON::ParserError
    # The body is not JSON; report it below.
  end
  raise ControlPlaneError, "Unexpected response: #{response.body.to_s[0, 200]}"
end

# The :- expansion treats an empty variable as unset. The password has no
# default; the Control Plane never returns it, so it comes from whoever
# created the database user.
base_url="${CONTROL_PLANE_URL:-http://localhost:3000}"
base_url="${base_url%/}"
database_id="${CONTROL_PLANE_DATABASE_ID:-cookbook}"
database_user="${CONTROL_PLANE_DATABASE_USER:-cookbook}"
database_password="${CONTROL_PLANE_DATABASE_PASSWORD:-}"

timeout_seconds=30

# The setting to change, and its new value.
setting=work_mem
value=16MB

# How long to wait for an update to finish, and how often to check.
wait_seconds=600
poll_seconds=5

# Sends a request and prints the reply, followed by the HTTP status on a
# line of its own; the status is 000 if no reply arrived. Any further
# arguments go to curl, such as --data @- to send standard input as the
# body.
request() {
    local method="$1" path="$2"
    shift 2
    curl --silent --show-error --max-time "$timeout_seconds" --request "$method" \
        --write-out '\n%{http_code}' "$@" "$base_url$path" || true
}

# Fetches a path from the server and prints the reply, or reports the
# error and fails.
get() {
    local reply status
    reply="$(request GET "$1")"
    status="${reply##*$'\n'}"
    if [[ "$status" != 200 ]]; then
        echo "Control Plane error: HTTP $status, ${reply%$'\n'*}" >&2
        return 1
    fi
    printf '%s\n' "${reply%$'\n'*}"
}

def update_database(client: httpx.Client, spec: dict) -> str:
    """Sends a whole database spec as an update, and returns the ID of the
    task that applies it. The Control Plane compares the spec with the
    current one and changes only what differs; the work happens in the
    task after the reply."""
    reply = call(client, "POST", f"/v1/databases/{DATABASE_ID}", {"spec": spec})
    return reply["task"]["task_id"]


def changed_spec(spec: dict) -> dict:
    """Returns a copy of the spec with the setting changed.
    postgresql_conf holds the Postgres settings that the Control Plane
    manages, and may be missing."""
    changed = copy.deepcopy(spec)
    changed.setdefault("postgresql_conf", {})[SETTING] = VALUE
    return changed

// The spec is kept as plain JSON, so that the update sends back every
// field that the server returned, including any that this program does
// not know about.
type Spec = Record<string, unknown> & { postgresql_conf?: Record<string, unknown> };

// Sends a whole database spec as an update, and returns the ID of the
// task that applies it. The Control Plane compares the spec with the
// current one and changes only what differs; the work happens in the task
// after the reply.
async function updateDatabase(spec: Spec, signal?: AbortSignal): Promise<string> {
  const reply = await call<{ task: { task_id: string } }>(
    "POST",
    `/v1/databases/${databaseId}`,
    { spec },
    signal,
  );
  return reply.task.task_id;
}

// Returns a copy of the spec with the setting changed. postgresql_conf
// holds the Postgres settings that the Control Plane manages, and may be
// missing.
function changedSpec(spec: Spec): Spec {
  return { ...spec, postgresql_conf: { ...spec.postgresql_conf, [setting]: value } };
}

/**
 * Sends a whole database spec as an update, and returns the ID of the
 * task that applies it. The Control Plane compares the spec with the
 * current one and changes only what differs; the work happens in the
 * task after the reply. The spec is kept as a JSON tree, so that the
 * update sends back every field that the server returned.
 */
static String updateDatabase(JsonNode spec)
        throws IOException, InterruptedException, ControlPlaneException {
    JsonNode reply = call("POST", "/v1/databases/" + DATABASE_ID, Map.of("spec", spec));
    String taskId = reply.path("task").path("task_id").asString("");
    if (taskId.isEmpty()) {
        throw new ControlPlaneException("Unexpected response: missing task_id");
    }
    return taskId;
}

/**
 * Returns a copy of the spec with the setting changed. postgresql_conf
 * holds the Postgres settings that the Control Plane manages, and may
 * be missing.
 */
static JsonNode changedSpec(JsonNode spec) {
    ObjectNode changed = (ObjectNode) spec.deepCopy();
    if (!changed.path("postgresql_conf").isObject()) {
        changed.putObject("postgresql_conf");
    }
    ((ObjectNode) changed.get("postgresql_conf")).put(SETTING, VALUE);
    return changed;
}

// Sends a whole database spec as an update, and returns the ID of the task
// that applies it. The Control Plane compares the spec with the current
// one and changes only what differs; the work happens in the task after
// the reply. The spec is kept as a JSON tree, so that the update sends back
// every field that the server returned.
static async Task<string> UpdateDatabaseAsync(
    HttpClient client, JsonObject spec, CancellationToken token)
{
    var body = new JsonObject { ["spec"] = spec.DeepClone() };
    var reply = await CallAsync(
        client, HttpMethod.Post, $"/v1/databases/{DatabaseId()}", body, token);
    return reply["task"]?["task_id"]?.ToString()
        ?? throw new ControlPlaneException("Unexpected response: missing task_id");
}

// Returns a copy of the spec with the setting changed. postgresql_conf
// holds the Postgres settings that the Control Plane manages, and may be
// missing.
static JsonObject ChangedSpec(JsonObject spec)
{
    var changed = spec.DeepClone().AsObject();
    if (changed["postgresql_conf"] is not JsonObject conf)
    {
        conf = new JsonObject();
        changed["postgresql_conf"] = conf;
    }
    conf[Setting] = Value;
    return changed;
}

// spec is kept as plain JSON, so that the update sends back every field
// that the server returned, including any that this program does not
// know about.
type spec = map[string]any

// updateDatabase sends a whole database spec as an update, and returns
// the ID of the task that applies it. The Control Plane compares the spec
// with the current one and changes only what differs; the work happens in
// the task after the reply.
func updateDatabase(ctx context.Context, s spec) (string, error) {
    var reply struct {
        Task struct {
            TaskID string `json:"task_id"`
        } `json:"task"`
    }
    err := call(ctx, http.MethodPost, "/v1/databases/"+databaseID, map[string]any{"spec": s}, &reply)
    return reply.Task.TaskID, err
}

// changedSpec returns a copy of the spec with the setting changed.
// postgresql_conf holds the Postgres settings that the Control Plane
// manages, and may be missing.
func changedSpec(s spec) spec {
    conf := map[string]any{}
    if current, ok := s["postgresql_conf"].(map[string]any); ok {
        maps.Copy(conf, current)
    }
    conf[setting] = value
    changed := maps.Clone(s)
    changed["postgresql_conf"] = conf
    return changed
}

/// Sends a whole database spec as an update, and returns the ID of the
/// task that applies it. The Control Plane compares the spec with the
/// current one and changes only what differs; the work happens in the
/// task after the reply.
fn update_database(client: &Client, spec: &Value) -> Result<String, BoxError> {
    let reply = call(
        client,
        Method::POST,
        &format!("/v1/databases/{}", database_id()),
        Some(&json!({ "spec": spec })),
    )?;
    reply["task"]["task_id"]
        .as_str()
        .map(str::to_string)
        .ok_or_else(|| "Unexpected response: missing task_id".into())
}

/// Returns a copy of the spec with the setting changed. postgresql_conf
/// holds the Postgres settings that the Control Plane manages, and may be
/// missing.
fn changed_spec(spec: &Value) -> Value {
    let mut changed = spec.clone();
    if !changed["postgresql_conf"].is_object() {
        changed["postgresql_conf"] = json!({});
    }
    changed["postgresql_conf"][SETTING] = json!(VALUE);
    changed
}

# Sends a whole database spec as an update, and returns the ID of the task
# that applies it. The Control Plane compares the spec with the current
# one and changes only what differs; the work happens in the task after
# the reply.
def update_database(spec)
  reply = call("POST", "/v1/databases/#{DATABASE_ID}", { "spec" => spec })
  reply.dig("task", "task_id") or
    raise ControlPlaneError, "Unexpected response: missing task_id"
end

# Returns a copy of the spec with the setting changed. postgresql_conf
# holds the Postgres settings that the Control Plane manages, and may be
# missing.
def changed_spec(spec)
  spec.merge("postgresql_conf" => (spec["postgresql_conf"] || {}).merge(SETTING => VALUE))
end

# Sends a whole database spec as an update, and prints the reply followed
# by the HTTP status, as request() does. The Control Plane compares the
# spec with the current one and changes only what differs; the work
# happens in a task after the reply.
update_database() {
    jq '{spec: .}' <<<"$1" |
        request POST "/v1/databases/$database_id" \
            --header "Content-Type: application/json" --data @-
}

# Prints the spec with the setting changed. postgresql_conf holds the
# Postgres settings that the Control Plane manages, and may be missing.
changed_spec() {
    jq --arg name "$setting" --arg value "$value" \
        '.postgresql_conf = ((.postgresql_conf // {}) + {($name): $value})' <<<"$1"
}

def is_transient(error: Exception) -> bool:
    """Reports whether a failed request is worth repeating: no reply
    arrived, the server failed, or it answered 409, which the Control
    Plane does whilst another task is still running on the database."""
    if isinstance(error, httpx.TransportError):
        return True
    status = getattr(error, "status", None)
    return status is not None and (status == 409 or status >= 500)


def wait_for_task(client: httpx.Client, task_id: str) -> None:
    """Polls a task until it finishes, and raises an error unless it
    completed. A failed poll is repeated if it is worth repeating."""
    deadline = time.monotonic() + WAIT_SECONDS
    while time.monotonic() < deadline:
        try:
            task = call(client, "GET", f"/v1/databases/{DATABASE_ID}/tasks/{task_id}")
        except (httpx.TransportError, ControlPlaneError) as e:
            if not is_transient(e):
                raise
            time.sleep(POLL_SECONDS)
            continue
        status = task.get("status")
        if status == "completed":
            return
        if status in ("failed", "canceled"):
            raise ControlPlaneError(f"the task finished with the status {status}")
        time.sleep(POLL_SECONDS)
    raise ControlPlaneError(f"the task was still running after {WAIT_SECONDS} seconds")

// Reports whether a failed request is worth repeating: no reply arrived,
// the server failed, or it answered 409, which the Control Plane does
// whilst another task, such as the update, is still running on the
// database.
function isTransient(err: unknown): boolean {
  if (!(err instanceof ControlPlaneError)) {
    return true;
  }
  return err.status !== undefined && (err.status === 409 || err.status >= 500);
}

// Polls a task until it finishes, and throws an error unless it
// completed. A failed poll is repeated if it is worth repeating. The
// cleanup passes a signal that never aborts.
async function waitForTask(taskId: string, signal: AbortSignal = interrupt.signal): Promise<void> {
  const deadline = Date.now() + waitMs;
  while (Date.now() < deadline) {
    let task: { status?: string } = {};
    try {
      task = await call<{ status?: string }>(
        "GET",
        `/v1/databases/${databaseId}/tasks/${taskId}`,
        undefined,
        signal,
      );
    } catch (err) {
      if (signal.aborted || !isTransient(err)) {
        throw err;
      }
    }
    if (task.status === "completed") {
      return;
    }
    if (task.status === "failed" || task.status === "canceled") {
      throw new ControlPlaneError(`the task finished with the status ${task.status}`);
    }
    await sleep(pollMs, undefined, { signal });
  }
  throw new ControlPlaneError(`the task was still running after ${waitMs / 1000} seconds`);
}

/**
 * Reports whether a failed request is worth repeating: no reply
 * arrived, the server failed, or it answered 409, which the Control
 * Plane does whilst another task, such as the update, is still running
 * on the database.
 */
static boolean isTransient(Exception e) {
    if (e instanceof ControlPlaneException cpe) {
        return cpe.status == 409 || cpe.status >= 500;
    }
    return e instanceof IOException;
}

/**
 * Polls a task until it finishes, and throws unless it completed. A
 * failed poll is repeated if it is worth repeating.
 */
static void waitForTask(String taskId)
        throws IOException, InterruptedException, ControlPlaneException {
    String path = "/v1/databases/" + DATABASE_ID + "/tasks/" + taskId;
    long deadline = System.nanoTime() + WAIT_FOR.toNanos();
    while (System.nanoTime() < deadline) {
        String status = "";
        try {
            status = call("GET", path, null).path("status").asString("");
        } catch (IOException | ControlPlaneException e) {
            if (!isTransient(e)) {
                throw e;
            }
        }
        if (status.equals("completed")) {
            return;
        }
        if (status.equals("failed") || status.equals("canceled")) {
            throw new ControlPlaneException(
                    "the task finished with the status " + status);
        }
        Thread.sleep(POLL_INTERVAL.toMillis());
    }
    throw new ControlPlaneException("the task was still running after "
            + WAIT_FOR.toSeconds() + " seconds");
}

// Reports whether a failed request is worth repeating: no reply arrived,
// the server failed, or it answered 409, which the Control Plane does
// whilst another task, such as the update, is still running on the
// database. TaskCanceledException is a timeout unless the caller's token
// was cancelled, which the caller checks.
static bool IsTransient(Exception e) => e switch
{
    ControlPlaneException cpe => cpe.Status is 409 or >= 500,
    HttpRequestException or TaskCanceledException => true,
    _ => false,
};

// Polls a task until it finishes, and throws an exception unless it
// completed. A failed poll is repeated if it is worth repeating.
static async Task WaitForTaskAsync(HttpClient client, string taskId, CancellationToken token)
{
    var clock = Stopwatch.StartNew();
    while (clock.Elapsed.TotalSeconds < WaitSeconds)
    {
        string? status = null;
        try
        {
            var task = await CallAsync(
                client, HttpMethod.Get, $"/v1/databases/{DatabaseId()}/tasks/{taskId}", null, token);
            status = task["status"]?.ToString();
        }
        catch (Exception e) when (IsTransient(e) && !token.IsCancellationRequested)
        {
            // Check again after the pause.
        }
        if (status == "completed")
        {
            return;
        }
        if (status is "failed" or "canceled")
        {
            throw new ControlPlaneException($"the task finished with the status {status}");
        }
        await Task.Delay(TimeSpan.FromSeconds(PollSeconds), token);
    }
    throw new ControlPlaneException($"the task was still running after {WaitSeconds} seconds");
}

// isTransient reports whether a failed request is worth repeating: no
// reply arrived, the server failed, or it answered 409, which the Control
// Plane does whilst another task is still running on the database.
func isTransient(err error) bool {
    status := statusOf(err)
    return status == 0 || status == http.StatusConflict || status >= 500
}

// waitForTask polls a task until it finishes, and returns an error unless
// it completed. A failed poll is repeated if it is worth repeating.
func waitForTask(ctx context.Context, taskID string) error {
    deadline := time.Now().Add(waitFor)
    for time.Now().Before(deadline) {
        var task struct {
            Status string `json:"status"`
        }
        err := call(ctx, http.MethodGet, "/v1/databases/"+databaseID+"/tasks/"+taskID, nil, &task)
        if err != nil && (ctx.Err() != nil || !isTransient(err)) {
            return err
        }
        switch {
        case err != nil:
        case task.Status == "completed":
            return nil
        case task.Status == "failed" || task.Status == "canceled":
            return fmt.Errorf("the task finished with the status %s", task.Status)
        }
        if err := sleep(ctx); err != nil {
            return err
        }
    }
    return fmt.Errorf("the task was still running after %d seconds", int(waitFor.Seconds()))
}

/// Reports whether a failed request is worth repeating: no reply arrived,
/// the server failed, or it answered 409, which the Control Plane does
/// whilst another task, such as the update, is still running on the
/// database.
fn is_transient(error: &BoxError) -> bool {
    match status_of(error) {
        Some(status) => status == 409 || status >= 500,
        None => error.is::<reqwest::Error>(),
    }
}

/// Polls a task until it finishes, and fails unless it completed. A failed
/// poll is repeated if it is worth repeating. When check is true, Ctrl-C
/// ends the wait.
fn wait_for_task(client: &Client, task_id: &str, check: bool) -> Result<(), BoxError> {
    let path = format!("/v1/databases/{}/tasks/{task_id}", database_id());
    let deadline = Instant::now() + WAIT_FOR;
    while Instant::now() < deadline {
        let task = call(client, Method::GET, &path, None);
        check_interrupted(check)?;
        match task {
            Ok(task) => match task["status"].as_str() {
                Some("completed") => return Ok(()),
                Some(status @ ("failed" | "canceled")) => {
                    return Err(format!("the task finished with the status {status}").into())
                }
                _ => {}
            },
            Err(e) if !is_transient(&e) => return Err(e),
            Err(_) => {}
        }
        pause(check)?;
    }
    Err(format!("the task was still running after {} seconds", WAIT_FOR.as_secs()).into())
}

# Reports whether a failed request is worth repeating: no reply arrived,
# the server failed, or it answered 409, which the Control Plane does
# whilst another task is still running on the database.
def transient?(error)
  return true unless error.is_a?(ControlPlaneError)

  !error.status.nil? && (error.status == 409 || error.status >= 500)
end

# Polls a task until it finishes, and raises an error unless it completed.
# A failed poll is repeated if it is worth repeating.
def wait_for_task(task_id)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + WAIT_SECONDS
  while Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline
    begin
      status = call("GET", "/v1/databases/#{DATABASE_ID}/tasks/#{task_id}")["status"]
    rescue ControlPlaneError, *NETWORK_ERRORS => e
      raise unless transient?(e)

      sleep POLL_SECONDS
      next
    end
    return if status == "completed"
    if %w[failed canceled].include?(status)
      raise ControlPlaneError, "the task finished with the status #{status}"
    end

    sleep POLL_SECONDS
  end
  raise ControlPlaneError, "the task was still running after #{WAIT_SECONDS} seconds"
end