Shared Node client

export RETRIEVER_API_URL="https://api.retriever.run"
export RETRIEVER_API_KEY="ret_live_..."

// retriever.mjs — Node 18+
export const API = process.env.RETRIEVER_API_URL;
export const KEY = process.env.RETRIEVER_API_KEY;
export const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export class RetrieverError extends Error {
  constructor(status, code, message, retryAfter) {
    super(`${status} ${code}: ${message}`);
    Object.assign(this, { status, code, retryAfter });
  }
}

// A failure that says nothing about the request: the network, a timeout, a
// body cut off mid-read, a 429 or a 5xx. Retrying it is safe for a GET, or a
// POST with an Idempotency-Key.
export function isTransient(error) {
  if (error instanceof RetrieverError) return error.status === 429 || error.status >= 500;
  return error.name === "TimeoutError" || (error instanceof TypeError && /fetch failed|terminated/.test(error.message));
}

// Body limits: 25 MiB on a table import, 10 MiB everywhere else.
const MAX_BODY_BYTES = 10 * 1024 * 1024;
const MAX_IMPORT_BODY_BYTES = 25 * 1024 * 1024;

export async function call(path, { method = "GET", body, headers = {}, retries = 5, timeoutMs = 60_000 } = {}) {
  const canRetryFailure = method === "GET" || "Idempotency-Key" in headers;
  // An oversized body never succeeds: refuse it here, never send or retry it.
  const requestBody = body !== undefined ? JSON.stringify(body) : undefined;
  const maxBytes = /^\/v1\/workspaces\/[^/]+\/tables$/.test(path) ? MAX_IMPORT_BODY_BYTES : MAX_BODY_BYTES;
  if (requestBody !== undefined && Buffer.byteLength(requestBody) > maxBytes) {
    throw new RetrieverError(413, "body_too_large", `The body is over ${maxBytes} bytes: split it.`);
  }
  for (let attempt = 0; ; attempt++) {
    let res;
    try {
      res = await fetch(`${API}${path}`, {
        method,
        signal: AbortSignal.timeout(timeoutMs),
        headers: {
          Authorization: `Bearer ${KEY}`,
          ...(body !== undefined ? { "Content-Type": "application/json" } : {}),
          ...headers,
        },
        body: requestBody,
      });
    } catch (networkError) {
      if (!canRetryFailure || attempt >= retries) throw networkError;
      await sleep(Math.min(30_000, 500 * 2 ** attempt));
      continue;
    }
    if (res.ok) {
      try {
        return await res.json();
      } catch (readError) {
        // A body cut off mid-read is retried; a body that is not JSON is not.
        if (!canRetryFailure || attempt >= retries || !isTransient(readError)) throw readError;
        await sleep(Math.min(30_000, 500 * 2 ** attempt));
        continue;
      }
    }

    const payload = await res.json().catch(() => ({}));
    const error = payload.error ?? {};
    const retryAfter = Number(res.headers.get("retry-after")) || undefined;
    // 413 body_too_large is never retried: the same body is refused again.
    const retryable = res.status === 429 || (res.status >= 500 && canRetryFailure);
    if (!retryable || attempt >= retries) {
      throw new RetrieverError(res.status, error.code ?? `http_${res.status}`, error.message ?? res.statusText, retryAfter);
    }
    await sleep(retryAfter ? retryAfter * 1000 : Math.min(30_000, 500 * 2 ** attempt));
  }
}

export async function download(path) {
  const absolute = path.startsWith("http://") || path.startsWith("https://");
  const res = await fetch(absolute ? path : `${API}${path}`, {
    headers: absolute ? {} : { Authorization: `Bearer ${KEY}` },
  });
  if (!res.ok) throw new RetrieverError(res.status, `http_${res.status}`, "download failed");
  return Buffer.from(await res.arrayBuffer());
}

// `timeoutMs` counts from the first poll that sees the run out of the queue.
// Time spent `queued` behind your organization's other runs does not count:
// a queued run starts on its own, so never create it again.
// A failed read (network, 429, 5xx, for example during a deploy) says nothing
// about the run: waitRun polls again with backoff, and only failures that
// last `timeoutMs` in a row, counted from the first one, end the wait. A 401
// or 404 (wrong key or run id) is thrown at once.
export async function waitRun(id, { timeoutMs = 60 * 60_000 } = {}) {
  let deadline = null;
  let failingSince = null;
  for (let failures = 0; ; ) {
    let run;
    try {
      // retries: 0 — waitRun owns the retries, so failingSince starts at the first failure.
      run = await call(`/v1/runs/${id}`, { retries: 0 });
      failures = 0;
      failingSince = null;
    } catch (error) {
      if (!isTransient(error)) throw error;
      failingSince ??= Date.now();
      if (Date.now() - failingSince >= timeoutMs) {
        throw new Error(`Could not read run ${id} for ${timeoutMs} ms (${error.message}). It continues on the server: call waitRun("${id}") again to resume.`, { cause: error });
      }
      await sleep(error.retryAfter ? Math.min(60_000, error.retryAfter * 1000) : Math.min(60_000, 1000 * 2 ** failures++));
      continue;
    }
    if (!["queued", "running", "recovering"].includes(run.status)) return run;
    if (run.status !== "queued") deadline ??= Date.now() + timeoutMs;
    if (deadline !== null && Date.now() > deadline) {
      throw new Error(`Run ${id} still active ${timeoutMs} ms after it started. It continues on the server: call waitRun("${id}") again to resume.`);
    }
    await sleep(15_000);
  }
}

export async function* paginateRows(rowsPath) {
  let cursor = null;
  do {
    const qs = new URLSearchParams({ limit: "1000", ...(cursor ? { cursor } : {}) });
    const page = await call(`${rowsPath}?${qs}`);
    yield* page.items;
    cursor = page.next_cursor;
  } while (cursor);
}