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>

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

export class Upstream {
  #config: HubConfig
  #fetch: typeof fetch
  #cache = new Map<string, Cached>()
  #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

    const response = await this.#exec(server, path, new URLSearchParams(sorted), token, { method: 'GET' }, this.#config.timeoutMs)
    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
  }

  /*
  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. `body` is a multipart upload (the app zip).
  */
  async write(server: HubServer, path: string, params: Params = {}, body?: FormData | URLSearchParams): Promise<unknown> {
    if (server.adminToken === undefined) throw new UpstreamError('rejected', 'no admin token configured')
    try {
      const init: RequestInit = body === undefined ? { method: 'POST' } : { method: 'POST', body }
      return await this.#exec(server, path, new URLSearchParams(params), server.adminToken, init, Math.max(this.#config.timeoutMs, WRITE_TIMEOUT_MS))
    } finally {
      this.invalidate(server.id)
    }
  }

  /** 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' },
      })
      envelope = (await res.json()) as typeof envelope
    } catch (e) {
      throw new UpstreamError('unreachable', this.#describe(e, [server.token, token], timeoutMs))
    } finally {
      clearTimeout(timer)
    }

    if (envelope?.status !== 'ok') {
      const why =
        envelope?.status === 'invalid-token'
          ? 'its API token was refused'
          : (envelope?.errorMessage ?? `status ${String(envelope?.status)}`)
      throw new UpstreamError('rejected', why)
    }
    return envelope.response
  }

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