Connecting to a Starfleet Database

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

{
  "name": "starfleet-connect-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 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. -->
  <PackageReference Include="Npgsql" Version="10.0.3" />
</ItemGroup>

// pgx is a pure Go driver and toolkit for PostgreSQL. The program calls
// the Starfleet API with the standard library alone.
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 of reqwest, with rustls for TLS and JSON bodies.
reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] }
serde_json = "1"
# The synchronous PostgreSQL client, with TLS through OpenSSL.
postgres = "0.19"
postgres-openssl = "0.5"
openssl = "0.10"
# The standard library cannot catch a signal, so this small crate does it.
signal-hook = "0.4"

source "https://rubygems.org"

# The program calls the API with Net::HTTP and JSON, which come with Ruby.
# The pg gem wraps libpq, so libpq and its headers must be installed.
gem "pg", "~> 1.6"

import os
import secrets
import signal
import sys
import time

import httpx
import psycopg

import { randomBytes } from "node:crypto";
import pg from "pg";
import process from "node:process";
import { setTimeout as sleep } from "node:timers/promises";

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.security.SecureRandom;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
import java.util.HexFormat;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.postgresql.util.PSQLException;
import tools.jackson.core.JacksonException;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;

using System.Globalization;
using System.Net;
using System.Net.Http.Headers;
using System.Runtime.InteropServices;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using Npgsql;

import (
    "bytes"
    "context"
    "crypto/rand"
    "encoding/hex"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "os"
    "os/signal"
    "strings"
    "syscall"
    "time"

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

use std::collections::hash_map::RandomState;
use std::env;
use std::error::Error;
use std::fmt;
use std::hash::{BuildHasher, Hasher};
use std::panic::{self, AssertUnwindSafe};
use std::process::ExitCode;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::LazyLock;
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::{Duration, Instant};

use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
use postgres::config::SslMode;
use postgres::Config;
use postgres_openssl::MakeTlsConnector;
use reqwest::blocking::Client;
use reqwest::Method;
use serde_json::{json, Value};
use signal_hook::consts::{SIGINT, SIGTERM};

require "json"
require "net/http"
require "pg"
require "securerandom"

# 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 the JSON replies and
# psql to connect to the database.
for tool in curl jq psql; do
    command -v "$tool" >/dev/null || { echo "Error: $tool is required" >&2; exit 1; }
done

# How long to wait for a database to become available, or to disappear
# after a delete, and how often to check.
TIMEOUT_SECONDS = 900
POLL_SECONDS = 10

# How many times to look for a database whose create reply was lost.
LOOKUP_ATTEMPTS = 3

# How many times to try connecting to the database before giving up.
CONNECT_ATTEMPTS = 6


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

    The or operator treats an empty variable as unset. The client ID and
    secret have no defaults, so a missing value becomes None, and the
    token request reports it.
    """
    return {
        "base_url": (
            os.environ.get("STARFLEET_API_URL") or "https://api.pgedge.com"
        ).rstrip("/"),
        "client_id": os.environ.get("STARFLEET_CLIENT_ID") or None,
        "client_secret": os.environ.get("STARFLEET_CLIENT_SECRET") or None,
    }

// How long to wait for a database to become available, or to disappear
// after a delete, and how often to check.
const TIMEOUT_MS = 900_000;
const POLL_MS = 10_000;

// How many times to look for a database whose create reply was lost.
const LOOKUP_ATTEMPTS = 3;

// How many times to try connecting to the database before giving up.
const CONNECT_ATTEMPTS = 6;

// Read the API settings from the environment. The || operator, unlike ??,
// also treats an empty variable as unset.
const baseUrl = (
  process.env.STARFLEET_API_URL || "https://api.pgedge.com"
).replace(/\/+$/, "");
const clientId = process.env.STARFLEET_CLIENT_ID || "";
const clientSecret = process.env.STARFLEET_CLIENT_SECRET || "";

// Ctrl-C and SIGTERM abort the signal below instead of ending the
// process, which lets the program delete its database 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());
}

/**
 * How long to wait for a database to become available, or to disappear
 * after a delete, and how often to check.
 */
static final int TIMEOUT_SECONDS = 900;
static final int POLL_SECONDS = 10;

/** How many times to look for a database whose create reply was lost. */
static final int LOOKUP_ATTEMPTS = 3;

/** How many times to try connecting to the database before giving up. */
static final int CONNECT_ATTEMPTS = 6;

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

/**
 * 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 API settings from the environment. The client ID and secret
 * have no defaults, so a missing value becomes null, and the token
 * request reports it.
 */
static Settings settings() {
    return new Settings(
            envOr("STARFLEET_API_URL", "https://api.pgedge.com")
                    .replaceAll("/+$", ""),
            envOr("STARFLEET_CLIENT_ID", null),
            envOr("STARFLEET_CLIENT_SECRET", null));
}

/**
 * Returns a deadline the given number of seconds from now, on a clock
 * that only moves forwards.
 */
static long deadline(int seconds) {
    return System.nanoTime() + TimeUnit.SECONDS.toNanos(seconds);
}

/** Reports whether a deadline is still in the future. */
static boolean before(long deadline) {
    return System.nanoTime() - deadline < 0;
}

/**
 * Guards the state that the clean-up uses, which the main thread and
 * the shutdown hook share (see main).
 */
static final Object LOCK = new Object();

/** Set by the shutdown hook when SIGINT or SIGTERM arrives. */
static volatile boolean interrupted;

/**
 * Throws InterruptedException once a signal has arrived, so that the
 * main thread stops making requests. A thread that holds the lock is
 * either creating the database or cleaning up, and both of those must
 * run to the end.
 */
static void checkInterrupted() throws InterruptedException {
    if (interrupted && !Thread.holdsLock(LOCK)) {
        throw new InterruptedException("interrupted");
    }
}

/** Waits for the given number of seconds, watching for a signal. */
static void pause(int seconds) throws InterruptedException {
    long end = deadline(seconds);
    while (before(end)) {
        checkInterrupted();
        Thread.sleep(200);
    }
}

// How long to wait for a database to become available, or to disappear
// after a delete, and how often to check.
const int TimeoutSeconds = 900;
const int PollSeconds = 10;

// How many times to look for a database whose create reply was lost.
const int LookupAttempts = 3;

// How many times to try connecting to the database before giving up.
const int ConnectAttempts = 6;

// 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 API settings from the environment. The client ID and secret
// have no defaults, so a missing value becomes null, and the token request
// reports it.
static (string BaseUrl, string? ClientId, string? ClientSecret) Settings() =>
    ((EnvValue("STARFLEET_API_URL") ?? "https://api.pgedge.com").TrimEnd('/'),
        EnvValue("STARFLEET_CLIENT_ID"),
        EnvValue("STARFLEET_CLIENT_SECRET"));

var cfg = Settings();

// Returns a deadline the given number of seconds from now, and reports
// whether it is still in the future, on a clock that only moves forwards.
static long Deadline(int seconds) => Environment.TickCount64 + seconds * 1000L;
static bool Before(long deadline) => Environment.TickCount64 < deadline;

// A token that SIGINT or SIGTERM cancels, which abandons the request or
// the wait in progress. The clean-up replaces it with CancellationToken.None,
// so that nothing can cut the clean-up short.
using var signalled = new CancellationTokenSource();
var cancel = signalled.Token;

// Catches SIGINT and SIGTERM. Setting Cancel stops the default action,
// which would end the program before it could delete the database, and a
// second signal does nothing more.
void OnSignal(PosixSignalContext context)
{
    context.Cancel = true;
    signalled.Cancel();
}
using var onInt = PosixSignalRegistration.Create(PosixSignal.SIGINT, OnSignal);
using var onTerm = PosixSignalRegistration.Create(PosixSignal.SIGTERM, OnSignal);

// Waits for the given number of seconds; a signal ends the wait early.
async Task Pause(int seconds) =>
    await Task.Delay(TimeSpan.FromSeconds(seconds), cancel);

const (
    // How long to wait for a database to become available, or to
    // disappear after a delete, and how often to check.
    timeout      = 15 * time.Minute
    pollInterval = 10 * time.Second

    // How many times to look for a database whose create reply was lost.
    lookupAttempts = 3

    // How many times to try connecting to the database before giving up.
    connectAttempts = 6
)

// 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
}

// baseURL is the root of the Starfleet API, without a trailing slash.
func baseURL() string {
    return strings.TrimRight(envOr("STARFLEET_API_URL", "https://api.pgedge.com"), "/")
}

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

/// How long to wait for a database to become available, or to disappear
/// after a delete, and how often to check.
const TIMEOUT: Duration = Duration::from_secs(900);
const POLL: Duration = Duration::from_secs(10);

/// How many times to look for a database whose create reply was lost.
const LOOKUP_ATTEMPTS: u32 = 3;

/// How many times to try connecting to the database before giving up.
const CONNECT_ATTEMPTS: u32 = 6;

/// 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,
    client_id: Option<String>,
    client_secret: Option<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 API settings from the environment. The client ID and secret
/// have no defaults, so a missing value becomes None, and the token
/// request reports it.
fn settings() -> Settings {
    Settings {
        base_url: env_value("STARFLEET_API_URL")
            .unwrap_or_else(|| "https://api.pgedge.com".to_string())
            .trim_end_matches('/')
            .to_string(),
        client_id: env_value("STARFLEET_CLIENT_ID"),
        client_secret: env_value("STARFLEET_CLIENT_SECRET"),
    }
}

/// Set when SIGINT or SIGTERM arrives.
static INTERRUPTED: AtomicBool = AtomicBool::new(false);

/// Set once the clean-up has started, which nothing may cut short.
static CLEANING: AtomicBool = AtomicBool::new(false);

/// Catches SIGINT and SIGTERM. The handler sets INTERRUPTED and replaces
/// the default action, which would end the program before it could delete
/// the database, so a second signal does nothing more.
fn catch_signals() -> Result<()> {
    for signal in [SIGINT, SIGTERM] {
        signal_hook::flag::register(signal, &INTERRUPTED)?;
    }
    Ok(())
}

/// The error that a request or a wait returns once a signal has arrived.
#[derive(Debug)]
struct Interrupted;

impl fmt::Display for Interrupted {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("interrupted")
    }
}

impl Error for Interrupted {}

/// Returns an error once a signal has arrived, so that the program stops
/// making requests, unless the clean-up has started.
fn check_interrupted() -> Result<()> {
    if INTERRUPTED.load(Ordering::SeqCst) && !CLEANING.load(Ordering::SeqCst) {
        return Err(Interrupted.into());
    }
    Ok(())
}

/// Waits for the given time, watching for a signal.
fn pause(duration: Duration) -> Result<()> {
    let end = Instant::now() + duration;
    while Instant::now() < end {
        check_interrupted()?;
        thread::sleep(Duration::from_millis(200));
    }
    Ok(())
}

# How long to wait for a database to become available, or to disappear
# after a delete, and how often to check.
TIMEOUT_SECONDS = 900
POLL_SECONDS = 10

# How many times to look for a database whose create reply was lost.
LOOKUP_ATTEMPTS = 3

# How many times to try connecting to the database before giving up.
CONNECT_ATTEMPTS = 6

# 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 API settings from the environment. The client ID and secret
# have no defaults, so a missing value becomes nil, and the token request
# reports it.
def settings
  {
    base_url: env_or("STARFLEET_API_URL", "https://api.pgedge.com").sub(%r{/+\z}, ""),
    client_id: env_or("STARFLEET_CLIENT_ID", nil),
    client_secret: env_or("STARFLEET_CLIENT_SECRET", nil)
  }
end

# The current time from a clock that only moves forwards, for deadlines.
def now
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

# How long to wait for a database to become available, or to disappear
# after a delete, and how often to check.
timeout_seconds=900
poll_seconds=10

# The :- expansion treats an empty variable as unset.
base_url="${STARFLEET_API_URL:-https://api.pgedge.com}"
base_url="${base_url%/}"
client_id="${STARFLEET_CLIENT_ID:-}"
client_secret="${STARFLEET_CLIENT_SECRET:-}"

def get_token(config: dict) -> str:
    """Exchanges the client ID and secret for an access token, which the
    Managed API accepts as a bearer token until it expires."""
    if not config["client_id"] or not config["client_secret"]:
        raise RuntimeError(
            "STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set"
        )

    response = httpx.post(
        f"{config['base_url']}/account/v1/oauth/token",
        json={
            "grant_type": "client_credentials",
            "client_id": config["client_id"],
            "client_secret": config["client_secret"],
        },
        timeout=30,
    )
    # Turn any 4xx or 5xx status into an httpx.HTTPStatusError.
    response.raise_for_status()
    return response.json()["access_token"]

// Exchanges the client ID and secret for an access token, which the
// Managed API accepts as a bearer token until it expires.
async function getToken(): Promise<string> {
  if (!clientId || !clientSecret) {
    throw new Error(
      "STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set",
    );
  }

  const response = await fetch(`${baseUrl}/account/v1/oauth/token`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      grant_type: "client_credentials",
      client_id: clientId,
      client_secret: clientSecret,
    }),
    signal: AbortSignal.timeout(30_000),
  });

  // fetch only rejects on network failures, so check the status here.
  if (!response.ok) {
    throw new Error(
      `token request failed: ${response.status} ${await response.text()}`,
    );
  }
  const body = (await response.json()) as { access_token: string };
  return body.access_token;
}

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

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

/** An error reply from the API, with its HTTP status. */
static class ApiException extends IOException {
    final int status;

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

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

/**
 * Sends a request with a JSON body, if there is one, and returns the
 * decoded JSON reply, or null if the reply has no body. A 4xx or 5xx
 * reply throws an ApiException.
 */
static JsonNode request(String method, String url, String path,
        String token, Object body) throws IOException, InterruptedException {
    checkInterrupted();
    HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(url))
            .timeout(Duration.ofSeconds(30));
    if (token != null) {
        request.header("Authorization", "Bearer " + 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 ApiException(response.statusCode(), method + " " + path
                + " failed with status " + response.statusCode() + ": "
                + response.body().strip());
    }
    String text = response.body();
    return text.isEmpty() ? null : JSON.readTree(text);
}

/**
 * Returns a field of a JSON object as text, and throws if the field is
 * missing.
 */
static String field(JsonNode node, String name) {
    JsonNode value = (node == null) ? null : node.get(name);
    if (value == null || value.isNull()) {
        throw new IllegalStateException("the reply has no " + name);
    }
    return value.asString();
}

/**
 * Exchanges the client ID and secret for an access token, which the
 * Managed API accepts as a bearer token until it expires.
 */
static String getToken(Settings cfg) throws IOException, InterruptedException {
    if (cfg.clientId() == null || cfg.clientSecret() == null) {
        throw new IllegalStateException(
                "STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set");
    }
    String path = "/account/v1/oauth/token";
    JsonNode token = request("POST", cfg.baseUrl() + path, path, null, Map.of(
            "grant_type", "client_credentials",
            "client_id", cfg.clientId(),
            "client_secret", cfg.clientSecret()));
    return field(token, "access_token");
}

// One client serves every request. Its timeout covers the whole exchange,
// the body of the reply included.
var http = new HttpClient { Timeout = TimeSpan.FromSeconds(30) };

// Sends a request with a JSON body, if there is one, and returns the
// decoded JSON reply, or null if the reply has no body. A 4xx or 5xx reply
// throws an HttpRequestException that carries the status.
async Task<JsonNode?> Request(string method, string url, string path,
    string? token, object? body, CancellationToken cancelRequest)
{
    using var request = new HttpRequestMessage(new HttpMethod(method), url);
    if (token is not null)
    {
        request.Headers.Authorization =
            new AuthenticationHeaderValue("Bearer", token);
    }
    if (body is not null)
    {
        // StringContent sends a Content-Length header; JsonContent would
        // stream the body in chunks, which not every server accepts.
        request.Content = new StringContent(
            JsonSerializer.Serialize(body), Encoding.UTF8, "application/json");
    }

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

// Returns a field of a JSON object as text, and throws if the field is
// missing.
static string Field(JsonNode? node, string name) =>
    node?[name] is JsonNode value
        ? value.ToString()
        : throw new InvalidOperationException($"the reply has no {name}");

// Exchanges the client ID and secret for an access token, which the
// Managed API accepts as a bearer token until it expires.
async Task<string> GetToken()
{
    if (cfg.ClientId is null || cfg.ClientSecret is null)
    {
        throw new InvalidOperationException(
            "STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set");
    }
    const string path = "/account/v1/oauth/token";
    var token = await Request("POST", cfg.BaseUrl + path, path, null, new
    {
        grant_type = "client_credentials",
        client_id = cfg.ClientId,
        client_secret = cfg.ClientSecret,
    }, cancel);
    return Field(token, "access_token");
}

// getToken exchanges the client ID and secret for an access token, which
// the Managed API accepts as a bearer token until it expires.
func getToken(ctx context.Context) (string, error) {
    clientID := os.Getenv("STARFLEET_CLIENT_ID")
    clientSecret := os.Getenv("STARFLEET_CLIENT_SECRET")
    if clientID == "" || clientSecret == "" {
        return "", errors.New("STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set")
    }

    body, err := json.Marshal(map[string]string{
        "grant_type":    "client_credentials",
        "client_id":     clientID,
        "client_secret": clientSecret,
    })
    if err != nil {
        return "", err
    }

    req, err := http.NewRequestWithContext(ctx, http.MethodPost,
        baseURL()+"/account/v1/oauth/token", bytes.NewReader(body))
    if err != nil {
        return "", err
    }
    req.Header.Set("Content-Type", "application/json")

    resp, err := httpClient.Do(req)
    if err != nil {
        return "", fmt.Errorf("token request failed: %w", err)
    }
    defer resp.Body.Close()

    // A non-2xx status is not a Go error, so check it explicitly.
    if resp.StatusCode != http.StatusOK {
        detail, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
        return "", fmt.Errorf("token request failed: %s %s", resp.Status, detail)
    }

    var token struct {
        AccessToken string `json:"access_token"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&token); err != nil {
        return "", fmt.Errorf("decoding the token response: %w", err)
    }
    return token.AccessToken, nil
}

/// One client serves every request. Its timeout covers the whole exchange,
/// the body of the reply included.
static CLIENT: LazyLock<Client> = LazyLock::new(|| {
    Client::builder()
        .timeout(Duration::from_secs(30))
        .build()
        .expect("could not create the HTTP client")
});

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

impl fmt::Display for ApiError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str(&self.message)
    }
}

impl Error for ApiError {}

/// Sends a request with a JSON body, if there is one, and returns the
/// decoded JSON reply, or null if the reply has no body. A 4xx or 5xx
/// reply returns an ApiError.
fn request(
    method: Method,
    url: &str,
    path: &str,
    token: Option<&str>,
    body: Option<&Value>,
) -> Result<Value> {
    let mut request = CLIENT.request(method.clone(), url);
    if let Some(token) = token {
        request = request.bearer_auth(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(ApiError {
            status,
            message: format!("{method} {path} failed with status {status}: {}", text.trim()),
        }
        .into());
    }
    if text.is_empty() {
        return Ok(Value::Null);
    }
    Ok(serde_json::from_str(&text)?)
}

/// Returns a field of a JSON object as text, and an error if the field is
/// missing.
fn field(value: &Value, name: &str) -> Result<String> {
    match value.get(name) {
        Some(Value::String(text)) => Ok(text.clone()),
        Some(Value::Null) | None => Err(format!("the reply has no {name}").into()),
        Some(other) => Ok(other.to_string()),
    }
}

/// Exchanges the client ID and secret for an access token, which the
/// Managed API accepts as a bearer token until it expires.
fn get_token(cfg: &Settings) -> Result<String> {
    let (Some(client_id), Some(client_secret)) = (&cfg.client_id, &cfg.client_secret) else {
        return Err("STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set".into());
    };
    check_interrupted()?;
    let path = "/account/v1/oauth/token";
    let body = json!({
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
    });
    let token = request(
        Method::POST,
        &format!("{}{path}", cfg.base_url),
        path,
        None,
        Some(&body),
    )?;
    field(&token, "access_token")
}

# An error reply from the API, with its HTTP status.
class ApiError < StandardError
  attr_reader :status

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

# Sends a request with a JSON body, if there is one, and returns the
# response. Net::HTTP has no overall timeout, so this limits both the
# connection and each read.
def send_request(method, url, headers, body)
  uri = URI(url)
  headers = headers.merge("Content-Type" => "application/json") if body
  Net::HTTP.start(uri.host, uri.port,
                  use_ssl: uri.scheme == "https",
                  open_timeout: 30, read_timeout: 30, write_timeout: 30) do |http|
    http.send_request(method, uri.request_uri, body&.to_json, headers)
  end
end

# Raises an ApiError for a 4xx or 5xx reply.
def check_status(method, path, response)
  return if response.code.to_i < 400

  raise ApiError.new(response.code.to_i,
                     "#{method} #{path} failed with status #{response.code}: " \
                     "#{response.body.to_s.strip}")
end

# Exchanges the client ID and secret for an access token, which the
# Managed API accepts as a bearer token until it expires.
def get_token(config)
  unless config[:client_id] && config[:client_secret]
    raise "STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set"
  end

  path = "/account/v1/oauth/token"
  response = send_request("POST", "#{config[:base_url]}#{path}", {}, {
                            grant_type: "client_credentials",
                            client_id: config[:client_id],
                            client_secret: config[:client_secret]
                          })
  check_status("POST", path, response)
  JSON.parse(response.body).fetch("access_token")
end

# Exchanges the client ID and secret for an access token, which the
# Managed API accepts as a bearer token until it expires, and prints
# the token.
get_token() {
    if [[ -z "$client_id" || -z "$client_secret" ]]; then
        echo "Error: STARFLEET_CLIENT_ID and STARFLEET_CLIENT_SECRET must be set" >&2
        return 1
    fi

    # jq builds the body from the environment, and curl reads it from
    # standard input with --data @-, so the secret never appears in the
    # process list. --fail-with-body makes curl exit non-zero on an HTTP
    # error status whilst still printing the API's error message.
    jq -n '{grant_type: "client_credentials",
            client_id: env.STARFLEET_CLIENT_ID,
            client_secret: env.STARFLEET_CLIENT_SECRET}' |
        curl --silent --show-error --fail-with-body --max-time 30 \
            --header "Content-Type: application/json" \
            --data @- \
            "$base_url/account/v1/oauth/token" |
        jq -er '.access_token'
}

def api(config: dict, token: str, method: str, path: str, body=None):
    """Sends an authenticated request to the Managed API and returns the
    decoded JSON reply, or None if the reply has no body."""
    response = httpx.request(
        method,
        f"{config['base_url']}/managed/v1{path}",
        headers={"Authorization": f"Bearer {token}"},
        json=body,
        timeout=30,
    )
    response.raise_for_status()
    return response.json() if response.content else None

// An error reply from the Managed API, with its HTTP status.
class ApiError extends Error {
  constructor(
    readonly status: number,
    message: string,
  ) {
    super(message);
  }
}

// Sends an authenticated request to the Managed API and returns the
// decoded JSON reply, or undefined if the reply has no body. The request
// stops when the user presses Ctrl-C, unless the caller passes a
// different signal, as the cleanup does.
async function api<T>(
  token: string,
  method: string,
  path: string,
  body?: unknown,
  signal: AbortSignal = interrupt.signal,
): Promise<T | undefined> {
  const response = await fetch(`${baseUrl}/managed/v1${path}`, {
    method,
    headers: {
      Authorization: `Bearer ${token}`,
      ...(body === undefined ? {} : { "Content-Type": "application/json" }),
    },
    body: body === undefined ? undefined : JSON.stringify(body),
    signal: AbortSignal.any([signal, AbortSignal.timeout(30_000)]),
  });
  if (!response.ok) {
    throw new ApiError(
      response.status,
      `${method} ${path} failed: ${response.status} ${await response.text()}`,
    );
  }
  const text = await response.text();
  return text ? (JSON.parse(text) as T) : undefined;
}

/**
 * Sends an authenticated request to the Managed API and returns the
 * decoded JSON reply, or null if the reply has no body.
 */
static JsonNode api(Settings cfg, String token, String method, String path,
        Object body) throws IOException, InterruptedException {
    return request(method, cfg.baseUrl() + "/managed/v1" + path, path,
            token, body);
}

// Sends an authenticated request to the Managed API and returns the
// decoded JSON reply, or null if the reply has no body. The create call
// passes interruptible: false, so that a signal cannot abandon a request
// whose result the program must not lose.
async Task<JsonNode?> Api(string token, string method, string path,
    object? body = null, bool interruptible = true) =>
    await Request(method, $"{cfg.BaseUrl}/managed/v1{path}", path, token,
        body, interruptible ? cancel : CancellationToken.None);

// APIError is an error reply from the Managed API, with its HTTP status.
type APIError struct {
    Status  int
    Message string
}

func (e *APIError) Error() string { return e.Message }

// api sends an authenticated request to the Managed API. It encodes body
// as JSON if it is not nil, and decodes the reply into result if that is
// not nil and the reply has content.
func api(ctx context.Context, token, method, path string, body, result 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.NewRequestWithContext(ctx, method, baseURL()+"/managed/v1"+path, reader)
    if err != nil {
        return err
    }
    req.Header.Set("Authorization", "Bearer "+token)
    if body != nil {
        req.Header.Set("Content-Type", "application/json")
    }

    resp, err := httpClient.Do(req)
    if err != nil {
        return fmt.Errorf("%s %s failed: %w", method, path, err)
    }
    defer resp.Body.Close()

    data, err := io.ReadAll(resp.Body)
    if err != nil {
        return err
    }
    if resp.StatusCode >= 300 {
        return &APIError{
            Status:  resp.StatusCode,
            Message: fmt.Sprintf("%s %s failed: %s %s", method, path, resp.Status, data),
        }
    }
    if result == nil || len(data) == 0 {
        return nil
    }
    return json.Unmarshal(data, result)
}

// sleep waits for the 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()
    }
}

/// Sends an authenticated request to the Managed API and returns the
/// decoded JSON reply, or null if the reply has no body. After a signal,
/// it returns an error without sending anything.
fn api(cfg: &Settings, token: &str, method: Method, path: &str, body: Option<&Value>) -> Result<Value> {
    check_interrupted()?;
    let url = format!("{}/managed/v1{path}", cfg.base_url);
    request(method, &url, path, Some(token), body)
}

# Sends an authenticated request to the Managed API and returns the
# decoded JSON reply, or nil if the reply has no body.
def api(config, token, method, path, body = nil)
  response = send_request(method, "#{config[:base_url]}/managed/v1#{path}",
                          { "Authorization" => "Bearer #{token}" }, body)
  check_status(method, path, response)
  text = response.body.to_s
  text.empty? ? nil : JSON.parse(text)
end

# Sends an authenticated request to the Managed API and prints the JSON
# reply. The first argument is the method, the second is the path, and any
# others go to curl, such as --data @- to send standard input as the body.
# The --config option reads the Authorization header from a file that
# bash creates from the printf command, so the token never appears in the
# process list; that leaves standard input free for the body.
api() {
    local method="$1" path="$2"
    shift 2
    curl --silent --show-error --fail-with-body --max-time 30 \
        --request "$method" \
        --config <(printf 'header = "Authorization: Bearer %s"\n' "$token") \
        "$@" "$base_url/managed/v1$path"
}

# Sends a request and prints only the HTTP status, for the calls that
# react to a particular status. A status of 000 means no reply arrived.
api_status() {
    curl --silent --output /dev/null --write-out '%{http_code}' --max-time 30 \
        --request "$1" \
        --config <(printf 'header = "Authorization: Bearer %s"\n' "$token") \
        "$base_url/managed/v1$2" || true
}

def choose_placement(config: dict, token: str) -> tuple[str, str]:
    """Returns the region and size for the new database.

    The API lists sizes smallest first, so the first one is the cheapest.
    The API does not price regions, so the example takes the first one.
    """
    regions = api(config, token, "GET", "/regions")
    sizes = api(config, token, "GET", "/sizes")
    if not regions or not sizes:
        raise RuntimeError("the API offers no region or size")
    return regions[0]["region"], sizes[0]["name"]

// Returns the region and size for the new database. The API lists sizes
// smallest first, so the first one is the cheapest. The API does not
// price regions, so the example takes the first one.
async function choosePlacement(
  token: string,
): Promise<{ region: string; size: string }> {
  const regions = await api<{ region: string }[]>(token, "GET", "/regions");
  const sizes = await api<{ name: string }[]>(token, "GET", "/sizes");
  if (!regions?.length || !sizes?.length) {
    throw new Error("the API offers no region or size");
  }
  return { region: regions[0].region, size: sizes[0].name };
}

/** The region and size of a new database. */
record Placement(String region, String size) {}

/**
 * Returns the region and size for the new database. The API lists sizes
 * smallest first, so the first one is the cheapest. The API does not
 * price regions, so the example takes the first one.
 */
static Placement choosePlacement(Settings cfg, String token)
        throws IOException, InterruptedException {
    JsonNode regions = api(cfg, token, "GET", "/regions", null);
    JsonNode sizes = api(cfg, token, "GET", "/sizes", null);
    if (regions == null || sizes == null || regions.size() == 0
            || sizes.size() == 0) {
        throw new IllegalStateException("the API offers no region or size");
    }
    return new Placement(field(regions.get(0), "region"),
            field(sizes.get(0), "name"));
}

// Returns the region and size for the new database. The API lists sizes
// smallest first, so the first one is the cheapest. The API does not price
// regions, so the example takes the first one.
async Task<(string Region, string Size)> ChoosePlacement(string token)
{
    var regions = await Api(token, "GET", "/regions") as JsonArray;
    var sizes = await Api(token, "GET", "/sizes") as JsonArray;
    if (regions is not { Count: > 0 } || sizes is not { Count: > 0 })
    {
        throw new InvalidOperationException("the API offers no region or size");
    }
    return (Field(regions[0], "region"), Field(sizes[0], "name"));
}

// choosePlacement returns the region and size for the new database. The
// API lists sizes smallest first, so the first one is the cheapest. The
// API does not price regions, so the example takes the first one.
func choosePlacement(ctx context.Context, token string) (region, size string, err error) {
    var regions []struct {
        Region string `json:"region"`
    }
    var sizes []struct {
        Name string `json:"name"`
    }
    if err = api(ctx, token, http.MethodGet, "/regions", nil, &regions); err != nil {
        return
    }
    if err = api(ctx, token, http.MethodGet, "/sizes", nil, &sizes); err != nil {
        return
    }
    if len(regions) == 0 || len(sizes) == 0 {
        err = errors.New("the API offers no region or size")
        return
    }
    return regions[0].Region, sizes[0].Name, nil
}

/// Returns the region and size for the new database. The API lists sizes
/// smallest first, so the first one is the cheapest. The API does not
/// price regions, so the example takes the first one.
fn choose_placement(cfg: &Settings, token: &str) -> Result<(String, String)> {
    let regions = api(cfg, token, Method::GET, "/regions", None)?;
    let sizes = api(cfg, token, Method::GET, "/sizes", None)?;
    match (regions.get(0), sizes.get(0)) {
        (Some(region), Some(size)) => Ok((field(region, "region")?, field(size, "name")?)),
        _ => Err("the API offers no region or size".into()),
    }
}

# Returns the region and size for the new database. The API lists sizes
# smallest first, so the first one is the cheapest. The API does not price
# regions, so the example takes the first one.
def choose_placement(config, token)
  regions = api(config, token, "GET", "/regions")
  sizes = api(config, token, "GET", "/sizes")
  raise "the API offers no region or size" if regions.to_a.empty? || sizes.to_a.empty?

  [regions[0].fetch("region"), sizes[0].fetch("name")]
end