technitium-console/hub/upstream.ts

/*
Calls to the real Technitium servers: `GET <url>/api/<path>?<params>&token=<token>`.
Successful answers are cached for a few seconds, so a dashboard that refreshes every
minute (and several people looking at it) does not load the servers.
*/
import type { HubConfig, HubServer } from './config.ts'

/*
`unreachable`: no answer (network, timeout, not JSON). `rejected`: the server answered
with a status other than `ok`. The hub never relays the server's `invalid-token`, or
the console would sign the hub's own user out because a hub token expired.
*/
export class UpstreamError extends Error {
  kind: 'unreachable' | 'rejected'
  constructor(kind: 'unreachable' | 'rejected', message: string) {
    super(message)
    this.name = 'UpstreamError'
    this.kind = kind
  }
}

export type Params = Record<string, string>

/* What `raw` hands back: the headers worth passing on and the body as a stream. */
export interface RawAnswer {
  contentType: string
  disposition?: string
  body: ReadableStream<Uint8Array> | null
  /* Stops the download (the caller went away). */
  abort: () => void
}

interface Cached {
  until: number
  response: unknown
}

const MAX_CACHED = 500

/* An app download or upload can take a while. */
const WRITE_TIMEOUT_MS = 60_000

/* How long reads of a server that timed out fail at once (the console polls every few seconds). */
const DOWN_MS = 10_000

/* How long a read may wait before it is asked again alongside. */
const HEDGE_MS = 2500

/* How long past its time a cached answer still stands in for a server that does not answer. */
const STALE_MS = 120_000

export class Upstream {
  #config: HubConfig
  #fetch: typeof fetch
  #cache = new Map<string, Cached>()
  #pending = new Map<string, Promise<unknown>>()
  #down = new Map<string, { until: number; error: UpstreamError }>()
  #now: () => number

  constructor(config: HubConfig, fetchImpl: typeof fetch = fetch, now: () => number = Date.now) {
    this.#config = config
    this.#fetch = fetchImpl
    this.#now = now
  }

  /** The `response` of the server's envelope, or an `UpstreamError`. */
  async call(server: HubServer, path: string, params: Params = {}, token: string = server.token): Promise<unknown> {
    const sorted = Object.entries(params).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
    const key = `${server.id}\n${path}\n${new URLSearchParams(sorted).toString()}`
    const hit = this.#cache.get(key)
    if (hit !== undefined && hit.until > this.#now()) return hit.response
    // A server that just timed out fails again at once, rather than holding every read for the whole timeout.
    // The last good answer, kept a little past its time: it stands in when the server does not answer.
    const stale = hit !== undefined && hit.until + STALE_MS > this.#now() ? hit : undefined
    const down = this.#down.get(server.id)
    if (down !== undefined && down.until > this.#now()) {
      if (stale !== undefined) return stale.response
      throw down.error
    }
    // The same read already on its way (the sidebar, the page and a panel ask together) is shared.
    const pending = this.#pending.get(key)
    if (pending !== undefined) return pending

    const call = this.#read(server, path, new URLSearchParams(sorted), token).then(
      (response) => {
        this.#down.delete(server.id)
        if (this.#config.cacheSeconds > 0) {
          if (this.#cache.size >= MAX_CACHED) this.#prune()
          this.#cache.set(key, { until: this.#now() + this.#config.cacheSeconds * 1000, response })
        }
        return response
      },
      (e: unknown) => {
        if (e instanceof UpstreamError && e.message.startsWith('timed out')) this.#down.set(server.id, { until: this.#now() + DOWN_MS, error: e })
        if (stale !== undefined && e instanceof UpstreamError && e.kind === 'unreachable') return stale.response
        throw e
      },
    )
    this.#pending.set(key, call)
    try {
      return await call
    } finally {
      this.#pending.delete(key)
    }
  }

  /*
  One read, asked a second time when the first is slow to answer: a request can stall on its own
  (a lost packet, a TLS handshake that hangs) while the server is fine, and a fresh one then answers
  at once. Whichever answers first wins; the first one's error is the error when both fail.
  shortcut: the losing request is left to finish or time out by itself, abort it if reads get heavy.
  */
  #read(server: HubServer, path: string, query: URLSearchParams, token: string): Promise<unknown> {
    const total = this.#config.timeoutMs
    const wait = Math.min(HEDGE_MS, total / 3)
    const first = this.#exec(server, path, new URLSearchParams(query), token, { method: 'GET' }, total)
    return new Promise((resolve, reject) => {
      let second: Promise<unknown> | undefined
      const timer = setTimeout(() => {
        second = this.#exec(server, path, new URLSearchParams(query), token, { method: 'GET' }, total - wait)
        second.then(resolve, () => {})
      }, wait)
      first.then(
        (r) => {
          clearTimeout(timer)
          resolve(r)
        },
        (e: unknown) => {
          clearTimeout(timer)
          if (second === undefined) reject(e)
          else second.then(resolve, () => reject(e))
        },
      )
    })
  }

  /*
  A write: POST with the server's ADMIN token and never the read one, never cached. Whatever
  the outcome, what the hub remembers of this server is dropped afterwards, so the next read
  shows the state the write left.

  The parameters travel in a form body (`application/x-www-form-urlencoded`) and only `token`
  stays in the query, so values (a record's data, a TSIG secret) do not end up in a proxy's
  access log. The server reads query or form alike. A `body` that is already a form gets the
  parameters added to it. Two bodies cannot be a form, so there the parameters stay in the
  query: a `FormData` (a multipart upload, the app zip or a zone file) and a string (a zone
  file as `text/plain`).
  */
  async write(server: HubServer, path: string, params: Params = {}, body?: FormData | URLSearchParams | string): Promise<unknown> {
    if (server.adminToken === undefined) throw new UpstreamError('rejected', 'no admin token configured')
    let query = new URLSearchParams()
    const init: RequestInit = { method: 'POST' }
    if (body === undefined || body instanceof URLSearchParams) {
      const form = new URLSearchParams(body)
      for (const [k, v] of Object.entries(params)) form.set(k, v)
      init.body = form
    } else {
      query = new URLSearchParams(params)
      init.body = body
      if (typeof body === 'string') init.headers = { 'content-type': 'text/plain; charset=utf-8' }
    }
    try {
      return await this.#exec(server, path, query, server.adminToken, init, Math.max(this.#config.timeoutMs, WRITE_TIMEOUT_MS))
    } finally {
      this.invalidate(server.id)
    }
  }

  /*
  A file the server answers with (a zone export, a log, a block list): read with the READ token,
  never cached. A JSON answer is the usual envelope, so an error is raised as for any call; any
  other answer is handed back as it is, to be streamed to the caller. The timeout covers the wait
  for the headers only: a big log may take longer to arrive.
  */
  async raw(server: HubServer, path: string, params: Params = {}): Promise<RawAnswer> {
    const query = new URLSearchParams(params)
    query.set('token', server.token)
    const controller = new AbortController()
    const timer = setTimeout(() => controller.abort(), this.#config.timeoutMs)
    let res: Response
    try {
      res = await this.#fetch(`${server.url}/api/${path}?${query.toString()}`, { method: 'GET', signal: controller.signal })
    } catch (e) {
      throw new UpstreamError('unreachable', this.#describe(e, [server.token], this.#config.timeoutMs))
    } finally {
      clearTimeout(timer)
    }
    const type = res.headers.get('content-type') ?? 'application/octet-stream'
    if (/json/i.test(type)) {
      let envelope: { status?: string; response?: unknown; errorMessage?: string }
      try {
        envelope = (await res.json()) as typeof envelope
      } catch (e) {
        throw new UpstreamError('unreachable', this.#describe(e, [server.token], this.#config.timeoutMs))
      }
      throw envelope?.status === 'ok' ? new UpstreamError('rejected', 'it sent JSON where a file was expected') : this.#rejection(envelope)
    }
    if (!res.ok) throw new UpstreamError('rejected', `status ${res.status}`)
    return {
      contentType: type,
      disposition: res.headers.get('content-disposition') ?? undefined,
      body: res.body,
      abort: () => controller.abort(),
    }
  }

  /** Forget every cached answer of one server. */
  invalidate(serverId: string): void {
    const prefix = `${serverId}\n`
    for (const key of this.#cache.keys()) if (key.startsWith(prefix)) this.#cache.delete(key)
  }

  async #exec(server: HubServer, path: string, query: URLSearchParams, token: string, init: RequestInit, timeoutMs: number): Promise<unknown> {
    query.set('token', token)
    const controller = new AbortController()
    const timer = setTimeout(() => controller.abort(), timeoutMs)

    let envelope: { status?: string; response?: unknown; errorMessage?: string }
    try {
      const res = await this.#fetch(`${server.url}/api/${path}?${query.toString()}`, {
        ...init,
        signal: controller.signal,
        headers: { accept: 'application/json', ...(init.headers as Record<string, string> | undefined) },
      })
      // A proxy in front of the server answers an outage with a page ("Bad Gateway"), not JSON.
      if (!res.ok && !/json/i.test(res.headers.get('content-type') ?? '')) {
        throw new UpstreamError('unreachable', `HTTP ${res.status} ${res.statusText}`.trim())
      }
      envelope = (await res.json()) as typeof envelope
    } catch (e) {
      if (e instanceof UpstreamError) throw e
      throw new UpstreamError('unreachable', this.#describe(e, [server.token, token], timeoutMs))
    } finally {
      clearTimeout(timer)
    }

    if (envelope?.status !== 'ok') throw this.#rejection(envelope)
    return envelope.response
  }

  /* The error for an envelope that is not `ok`. */
  #rejection(envelope: { status?: string; errorMessage?: string } | undefined): UpstreamError {
    const why =
      envelope?.status === 'invalid-token'
        ? 'its API token was refused'
        : (envelope?.errorMessage ?? `status ${String(envelope?.status)}`)
    return new UpstreamError('rejected', why)
  }

  #prune(): void {
    const now = this.#now()
    for (const [key, entry] of this.#cache) if (entry.until <= now) this.#cache.delete(key)
    // Still full of live entries: drop the oldest inserted.
    while (this.#cache.size >= MAX_CACHED) this.#cache.delete(this.#cache.keys().next().value as string)
  }

  /* A short reason without anything that could carry the token. */
  #describe(e: unknown, secrets: string[], timeoutMs: number): string {
    if (e instanceof Error && e.name === 'AbortError') return `timed out after ${timeoutMs} ms`
    const cause = e instanceof Error ? (e.cause as { code?: unknown } | undefined) : undefined
    const text = typeof cause?.code === 'string' ? cause.code : e instanceof Error ? e.message : 'request failed'
    return secrets.reduce((out, secret) => (secret === '' ? out : out.split(secret).join('***')), text)
  }
}