Creating a Database

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

{
  "name": "create-database",
  "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 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>
</dependencies>

<!-- HttpClient and System.Text.Json are part of .NET, so the project
     needs no packages. -->

// The program uses only the standard library, so there is nothing to add.

[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"
# Handling Ctrl-C, and with the termination feature SIGTERM, so that the
# program can delete its database before it exits.
ctrlc = { version = "3", features = ["termination"] }

source "https://rubygems.org"

# The program uses only Net::HTTP and json from the standard library, so
# there is no gem to add.

import os
import re
import secrets
import signal
import sys
import time

import httpx

// Node.js provides fetch and AbortSignal as globals; setTimeout from
// timers/promises returns a promise, which suits a polling loop.
import { randomBytes } from "node:crypto";
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.time.Duration;
import java.util.HexFormat;
import java.util.List;
import java.util.Map;
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;

using System.Diagnostics;
using System.Net.Http.Json;
using System.Runtime.InteropServices;
using System.Security.Cryptography;
using System.Text.Json;
using System.Text.RegularExpressions;
using System.Text.Json.Nodes;

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

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

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

require "json"
require "net/http"
require "securerandom"
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 and jq to read the JSON replies.
for tool in curl jq; 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 new database
# listens on the given port on its host, so choose one that is free.
BASE_URL = (os.environ.get("CONTROL_PLANE_URL") or "http://localhost:3000").rstrip("/")
PORT_TEXT = os.environ.get("CONTROL_PLANE_RECIPE_PORT") or "6433"

TIMEOUT_SECONDS = 30.0

# How long to wait for a database to become available, or to disappear
# after a delete, 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
// new database listens on the given port on its host, so choose one that
// is free.
const baseUrl = (process.env.CONTROL_PLANE_URL || "http://localhost:3000").replace(/\/+$/, "");
const portText = process.env.CONTROL_PLANE_RECIPE_PORT || "6433";

const timeoutMs = 30_000;

// How long to wait for a database to become available, or to disappear
// after a delete, 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 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());
}

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

// 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 new database listens on the given port on its host, so choose
// one that is free.
static final String BASE_URL =
        envOr("CONTROL_PLANE_URL", "http://localhost:3000")
                .replaceAll("/+$", "");
static final String PORT = envOr("CONTROL_PLANE_RECIPE_PORT", "6433");

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

// How long to wait for a database to become available, or to
// disappear after a delete, 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('/');

// The new database listens on the given port on its host, so choose one
// that is free.
static string PortText() => EnvOr("CONTROL_PLANE_RECIPE_PORT", "6433");

const int TimeoutSeconds = 30;

// How long to wait for a database to become available, or to disappear
// after a delete, 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 new database listens on the given port on its host, so choose one
// that is free.
var (
    baseURL  = strings.TrimRight(envOr("CONTROL_PLANE_URL", "http://localhost:3000"), "/")
    portText = envOr("CONTROL_PLANE_RECIPE_PORT", "6433")
)

// parsePort returns the port number in text, and false unless text is a
// plain decimal number from 1 to 65535.
func parsePort(text string) (int, bool) {
    number, err := strconv.Atoi(text)
    if err != nil || number < 1 || number > 65535 || strconv.Itoa(number) != text {
        return 0, false
    }
    return number, true
}

const 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 (
    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
}

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()
}

/// The port that the new database listens on, on its host; choose one
/// that is free.
fn port_text() -> String {
    env_or("CONTROL_PLANE_RECIPE_PORT", "6433")
}

/// Returns the port number in text, or None unless text is a plain
/// decimal number from 1 to 65535.
fn parse_port(text: &str) -> Option<u16> {
    let plain = !text.starts_with('0')
        && text.len() <= 5
        && text.bytes().all(|b| b.is_ascii_digit());
    if plain { text.parse().ok().filter(|&port| port > 0) } else { None }
}

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

// How long to wait for a database to become available, or to disappear
// after a delete, 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 delete its database 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 the user has pressed Ctrl-C. A request does not stop for
/// Ctrl-C, so the program checks the flag after each one.
fn check_interrupted() -> Result<(), BoxError> {
    if INTERRUPTED.load(Ordering::SeqCst) {
        return Err("interrupted".into());
    }
    Ok(())
}

/// Waits for the poll interval, and fails if the user presses Ctrl-C.
fn pause() -> Result<(), BoxError> {
    let end = Instant::now() + POLL_INTERVAL;
    while Instant::now() < end {
        check_interrupted()?;
        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 new database listens on the given port on its host, so choose one
# that is free.
BASE_URL = env_or("CONTROL_PLANE_URL", "http://localhost:3000").sub(%r{/+\z}, "")
PORT_TEXT = env_or("CONTROL_PLANE_RECIPE_PORT", "6433")

TIMEOUT_SECONDS = 30

# How long to wait for a database to become available, or to disappear
# after a delete, 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 new database
# listens on the given port on its host, so choose one that is free.
base_url="${CONTROL_PLANE_URL:-http://localhost:3000}"
base_url="${base_url%/}"
port="${CONTROL_PLANE_RECIPE_PORT:-6433}"

# How long to wait for a database to become available, or to disappear
# after a delete, 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 30 --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 first_host(client: httpx.Client) -> str:
    """Returns the ID of the first host in the cluster. A database places
    each of its nodes on one or more hosts, which run the Postgres
    containers."""
    hosts = call(client, "GET", "/v1/hosts").get("hosts") or []
    if not hosts:
        raise ControlPlaneError("the cluster has no hosts")
    return hosts[0]["id"]

// Returns the ID of the first host in the cluster. A database places each
// of its nodes on one or more hosts, which run the Postgres containers.
async function firstHost(): Promise<string> {
  const reply = await call<{ hosts?: { id: string }[] }>("GET", "/v1/hosts");
  const host = reply.hosts?.[0];
  if (!host) {
    throw new ControlPlaneError("the cluster has no hosts");
  }
  return host.id;
}

/**
 * Returns the ID of the first host in the cluster. A database places
 * each of its nodes on one or more hosts, which run the Postgres
 * containers.
 */
static String firstHost()
        throws IOException, InterruptedException, ControlPlaneException {
    JsonNode hosts = call("GET", "/v1/hosts", null).path("hosts");
    if (hosts.isEmpty()) {
        throw new ControlPlaneException("the cluster has no hosts");
    }
    return hosts.get(0).path("id").asString();
}

// Returns the ID of the first host in the cluster. A database places each
// of its nodes on one or more hosts, which run the Postgres containers.
static async Task<string> FirstHostAsync(HttpClient client, CancellationToken token)
{
    var reply = await CallAsync(client, HttpMethod.Get, "/v1/hosts", null, token);
    if (reply["hosts"] is not JsonArray { Count: > 0 } hosts ||
        hosts[0]?["id"]?.ToString() is not { } host)
    {
        throw new ControlPlaneException("the cluster has no hosts");
    }
    return host;
}

// firstHost returns the ID of the first host in the cluster. A database
// places each of its nodes on one or more hosts, which run the Postgres
// containers.
func firstHost(ctx context.Context) (string, error) {
    var reply struct {
        Hosts []struct {
            ID string `json:"id"`
        } `json:"hosts"`
    }
    if err := call(ctx, http.MethodGet, "/v1/hosts", nil, &reply); err != nil {
        return "", err
    }
    if len(reply.Hosts) == 0 {
        return "", errors.New("the cluster has no hosts")
    }
    return reply.Hosts[0].ID, nil
}

/// Returns the ID of the first host in the cluster. A database places each
/// of its nodes on one or more hosts, which run the Postgres containers.
fn first_host(client: &Client) -> Result<String, BoxError> {
    let reply = call(client, Method::GET, "/v1/hosts", None)?;
    reply["hosts"][0]["id"]
        .as_str()
        .map(str::to_string)
        .ok_or_else(|| "the cluster has no hosts".into())
}

# Returns the ID of the first host in the cluster. A database places each
# of its nodes on one or more hosts, which run the Postgres containers.
def first_host
  host = (call("GET", "/v1/hosts")["hosts"] || []).first
  raise ControlPlaneError, "the cluster has no hosts" unless host

  host["id"]
end

# Prints the ID of the first host in the cluster. A database places each of
# its nodes on one or more hosts, which run the Postgres containers.
first_host() {
    get /v1/hosts | jq -er '.hosts[0].id'
}

def create_database(client: httpx.Client, database_id: str, host: str) -> None:
    """Asks the Control Plane to create a database with one node. The reply
    holds the database and the task that creates it; the work happens in
    the background after the reply."""
    call(client, "POST", "/v1/databases", {
        "id": database_id,
        "spec": {
            "database_name": "recipe",
            "port": int(PORT_TEXT),
            "nodes": [{"name": "n1", "host_ids": [host]}],
        },
    })

// Asks the Control Plane to create a database with one node. The reply
// holds the database and the task that creates it; the work happens in the
// background after the reply.
async function createDatabase(databaseId: string, host: string): Promise<void> {
  await call("POST", "/v1/databases", {
    id: databaseId,
    spec: {
      database_name: "recipe",
      port: Number(portText),
      nodes: [{ name: "n1", host_ids: [host] }],
    },
  });
}

/**
 * Asks the Control Plane to create a database with one node. The reply
 * holds the database and the task that creates it; the work happens in
 * the background after the reply.
 */
static void createDatabase(String databaseId, String host)
        throws IOException, InterruptedException, ControlPlaneException {
    Map<String, Object> body = Map.of(
            "id", databaseId,
            "spec", Map.of(
                    "database_name", "recipe",
                    "port", Integer.parseInt(PORT),
                    "nodes", List.of(Map.of(
                            "name", "n1",
                            "host_ids", List.of(host)))));
    call("POST", "/v1/databases", body);
}

// Asks the Control Plane to create a database with one node. The reply
// holds the database and the task that creates it; the work happens in the
// background after the reply.
static async Task CreateDatabaseAsync(
    HttpClient client, string databaseId, string host, CancellationToken token)
{
    var body = new
    {
        id = databaseId,
        spec = new
        {
            database_name = "recipe",
            port = int.Parse(PortText()),
            nodes = new[] { new { name = "n1", host_ids = new[] { host } } },
        },
    };
    await CallAsync(client, HttpMethod.Post, "/v1/databases", body, token);
}

// createDatabase asks the Control Plane to create a database with one
// node. The reply holds the database and the task that creates it; the
// work happens in the background after the reply.
func createDatabase(ctx context.Context, databaseID, host string, port int) error {
    body := map[string]any{
        "id": databaseID,
        "spec": map[string]any{
            "database_name": "recipe",
            "port":          port,
            "nodes": []map[string]any{
                {"name": "n1", "host_ids": []string{host}},
            },
        },
    }
    return call(ctx, http.MethodPost, "/v1/databases", body, nil)
}

/// Asks the Control Plane to create a database with one node. The reply
/// holds the database and the task that creates it; the work happens in
/// the background after the reply.
fn create_database(
    client: &Client,
    database_id: &str,
    host: &str,
    port: u16,
) -> Result<(), BoxError> {
    let body = json!({
        "id": database_id,
        "spec": {
            "database_name": "recipe",
            "port": port,
            "nodes": [{"name": "n1", "host_ids": [host]}],
        },
    });
    call(client, Method::POST, "/v1/databases", Some(&body))?;
    Ok(())
}

# Asks the Control Plane to create a database with one node. The reply
# holds the database and the task that creates it; the work happens in the
# background after the reply.
def create_database(database_id, host)
  call("POST", "/v1/databases", {
         id: database_id,
         spec: {
           database_name: "recipe",
           port: Integer(PORT_TEXT, 10),
           nodes: [{ name: "n1", host_ids: [host] }]
         }
       })
end

# Asks the Control Plane to create a database with one node, and prints the
# reply followed by the HTTP status, as request() does. The reply holds the
# database and the task that creates it; the work happens in the
# background after the reply.
create_database() {
    local id="$1" host="$2"
    jq -n --arg id "$id" --arg host "$host" --argjson port "$port" \
        '{id: $id,
          spec: {database_name: "recipe",
                 port: $port,
                 nodes: [{name: "n1", host_ids: [$host]}]}}' |
        request POST /v1/databases \
            --header "Content-Type: application/json" --data @-
}

def wait_until_available(client: httpx.Client, database_id: str) -> int:
    """Polls the database, printing its state each time it changes, until
    the state is available, and returns the port of its first instance.
    Each instance runs one node, and its connection_info holds the address
    and port that clients use. A failed state ends the wait at once."""
    deadline = time.monotonic() + WAIT_SECONDS
    last = None
    while time.monotonic() < deadline:
        database = call(client, "GET", f"/v1/databases/{database_id}")
        state = database.get("state")
        if state != last:
            print(f"Database state: {state}")
            last = state
        if state == "available":
            instances = database.get("instances") or []
            if not instances:
                raise ControlPlaneError("the database is available but has no instances")
            return instances[0]["connection_info"]["port"]
        if state == "failed":
            raise ControlPlaneError("the database has the state failed")
        time.sleep(POLL_SECONDS)
    raise ControlPlaneError(
        f"the database was not available after {WAIT_SECONDS} seconds")

interface Database {
  state?: string;
  instances?: { connection_info: { port: number } }[];
}

// Polls the database, printing its state each time it changes, until the
// state is available, and returns the port of its first instance. Each
// instance runs one node, and its connection_info holds the address and
// port that clients use. A failed state ends the wait at once.
async function waitUntilAvailable(databaseId: string): Promise<number> {
  const deadline = Date.now() + waitMs;
  let last: string | undefined;
  while (Date.now() < deadline) {
    const database = await call<Database>("GET", `/v1/databases/${databaseId}`);
    if (database.state !== last) {
      console.log(`Database state: ${database.state}`);
      last = database.state;
    }
    if (database.state === "available") {
      const instance = database.instances?.[0];
      if (!instance) {
        throw new ControlPlaneError("the database is available but has no instances");
      }
      return instance.connection_info.port;
    }
    if (database.state === "failed") {
      throw new ControlPlaneError("the database has the state failed");
    }
    await sleep(pollMs, undefined, { signal: interrupt.signal });
  }
  throw new ControlPlaneError(`the database was not available after ${waitMs / 1000} seconds`);
}

/**
 * Polls the database, printing its state each time it changes, until
 * the state is available, and returns the port of its first instance.
 * Each instance runs one node, and its connection_info holds the
 * address and port that clients use. A failed state ends the wait at
 * once.
 */
static int waitUntilAvailable(String databaseId)
        throws IOException, InterruptedException, ControlPlaneException {
    long deadline = System.nanoTime() + WAIT_FOR.toNanos();
    String last = "";
    while (System.nanoTime() < deadline) {
        JsonNode database = call("GET", "/v1/databases/" + databaseId, null);
        String state = database.path("state").asString("");
        if (!state.equals(last)) {
            System.out.println("Database state: " + state);
            last = state;
        }
        if (state.equals("available")) {
            JsonNode instance = database.path("instances").path(0);
            if (instance.isMissingNode()) {
                throw new ControlPlaneException(
                        "the database is available but has no instances");
            }
            return instance.path("connection_info").path("port").asInt();
        }
        if (state.equals("failed")) {
            throw new ControlPlaneException("the database has the state failed");
        }
        Thread.sleep(POLL_INTERVAL.toMillis());
    }
    throw new ControlPlaneException("the database was not available after "
            + WAIT_FOR.toSeconds() + " seconds");
}

// Polls the database, printing its state each time it changes, until the
// state is available, and returns the port of its first instance. Each
// instance runs one node, and its connection_info holds the address and
// port that clients use. A failed state ends the wait at once.
static async Task<string> WaitUntilAvailableAsync(
    HttpClient client, string databaseId, CancellationToken token)
{
    var clock = Stopwatch.StartNew();
    string? last = null;
    while (clock.Elapsed.TotalSeconds < WaitSeconds)
    {
        var database = await CallAsync(
            client, HttpMethod.Get, $"/v1/databases/{databaseId}", null, token);
        var state = database["state"]?.ToString();
        if (state != last)
        {
            Console.WriteLine($"Database state: {state}");
            last = state;
        }
        if (state == "available")
        {
            if (database["instances"] is not JsonArray { Count: > 0 } instances)
            {
                throw new ControlPlaneException(
                    "the database is available but has no instances");
            }
            return instances[0]?["connection_info"]?["port"]?.ToString() ?? "";
        }
        if (state == "failed")
        {
            throw new ControlPlaneException("the database has the state failed");
        }
        await Task.Delay(TimeSpan.FromSeconds(PollSeconds), token);
    }
    throw new ControlPlaneException(
        $"the database was not available after {WaitSeconds} seconds");
}

// waitUntilAvailable polls the database, printing its state each time it
// changes, until the state is available, and returns the port of its
// first instance. Each instance runs one node, and its connection_info
// holds the address and port that clients use. A failed state ends the
// wait at once.
func waitUntilAvailable(ctx context.Context, databaseID string) (int, error) {
    deadline := time.Now().Add(waitFor)
    last := ""
    for time.Now().Before(deadline) {
        var database struct {
            State     string `json:"state"`
            Instances []struct {
                ConnectionInfo struct {
                    Port int `json:"port"`
                } `json:"connection_info"`
            } `json:"instances"`
        }
        err := call(ctx, http.MethodGet, "/v1/databases/"+databaseID, nil, &database)
        if err != nil {
            return 0, err
        }
        if database.State != last {
            fmt.Printf("Database state: %s\n", database.State)
            last = database.State
        }
        switch {
        case database.State == "available" && len(database.Instances) == 0:
            return 0, errors.New("the database is available but has no instances")
        case database.State == "available":
            return database.Instances[0].ConnectionInfo.Port, nil
        case database.State == "failed":
            return 0, errors.New("the database has the state failed")
        }
        if err := sleep(ctx); err != nil {
            return 0, err
        }
    }
    return 0, fmt.Errorf("the database was not available after %v", waitFor)
}

/// Polls the database, printing its state each time it changes, until the
/// state is available, and returns the port of its first instance. Each
/// instance runs one node, and its connection_info holds the address and
/// port that clients use. A failed state ends the wait at once.
fn wait_until_available(client: &Client, database_id: &str) -> Result<u64, BoxError> {
    let deadline = Instant::now() + WAIT_FOR;
    let mut last = String::new();
    while Instant::now() < deadline {
        let database = call(client, Method::GET, &format!("/v1/databases/{database_id}"), None)?;
        check_interrupted()?;
        let state = database["state"].as_str().unwrap_or("unknown").to_string();
        if state != last {
            println!("Database state: {state}");
            last = state.clone();
        }
        if state == "available" {
            let instance = database["instances"]
                .get(0)
                .ok_or("the database is available but has no instances")?;
            return instance["connection_info"]["port"]
                .as_u64()
                .ok_or_else(|| "Unexpected response: missing port".into());
        }
        if state == "failed" {
            return Err("the database has the state failed".into());
        }
        pause()?;
    }
    Err(format!("the database was not available after {} seconds", WAIT_FOR.as_secs()).into())
}

# Polls the database, printing its state each time it changes, until the
# state is available, and returns the port of its first instance. Each
# instance runs one node, and its connection_info holds the address and
# port that clients use. A failed state ends the wait at once.
def wait_until_available(database_id)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + WAIT_SECONDS
  last = nil
  while Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline
    database = call("GET", "/v1/databases/#{database_id}")
    state = database["state"]
    if state != last
      puts "Database state: #{state}"
      last = state
    end
    if state == "available"
      instance = (database["instances"] || []).first
      raise ControlPlaneError, "the database is available but has no instances" unless instance

      return instance.fetch("connection_info").fetch("port")
    end
    raise ControlPlaneError, "the database has the state failed" if state == "failed"

    sleep POLL_SECONDS
  end
  raise ControlPlaneError, "the database was not available after #{WAIT_SECONDS} seconds"
end

# Polls the database, printing its state each time it changes, until the
# state is available, and then sets available_port to the port of its
# first instance. Each instance runs one node, and its connection_info
# holds the address and port that clients use. A failed state ends the
# wait at once.
wait_until_available() {
    local id="$1" deadline=$((SECONDS + wait_seconds)) database state last=""
    while ((SECONDS < deadline)); do
        database="$(get "/v1/databases/$id")"
        state="$(jq -er '.state' <<<"$database")"
        if [[ "$state" != "$last" ]]; then
            echo "Database state: $state"
            last="$state"
        fi
        if [[ "$state" == "available" ]]; then
            available_port="$(jq -er '.instances[0].connection_info.port' <<<"$database")" || {
                echo "Control Plane error: the database is available but has no instances" >&2
                return 1
            }
            return 0
        fi
        if [[ "$state" == "failed" ]]; then
            echo "Control Plane error: the database has the state failed" >&2
            return 1
        fi
        sleep "$poll_seconds"
    done
    echo "Control Plane error: the database was not available after $wait_seconds seconds" >&2
    return 1
}