import { ApiError } from "../errors.js"; import { isSafeToReplay, resolveRetryOptions, withRetry, type ResolvedRetryOptions, type RetryOptions, } from "../retry.js"; // `ApiError` is defined in `src/errors.ts` so that the retry classifier can // recognise it without importing this module, which imports the classifier. // It is re-exported here because this is where callers have always imported it // from, and it must remain one class: a second copy would make `instanceof` // fail in the classifier and every 5xx would look permanent. export { ApiError }; const DEFAULT_API_ORIGIN = "https://api.ente.io"; const DEFAULT_FILES_ORIGIN = "https://files.ente.io"; const DEFAULT_THUMBS_ORIGIN = "https://thumbnails.ente.io"; const CLIENT_PACKAGE = "berlin.sneak.quak"; // Two deadlines of different kinds. `requestTimeoutMs` bounds a whole JSON // call. `downloadTimeoutMs` is an idle deadline: a file or thumbnail download // is aborted only when no bytes have arrived for that long, so a large video on // a slow link that keeps making progress is never cut off. export const DEFAULT_REQUEST_TIMEOUT_MS = 30_000; export const DEFAULT_DOWNLOAD_TIMEOUT_MS = 60_000; export interface ApiClientOptions { apiOrigin?: string; filesOrigin?: string; thumbsOrigin?: string; authToken?: string; fetch?: typeof globalThis.fetch; userAgent?: string; retry?: RetryOptions; requestTimeoutMs?: number; downloadTimeoutMs?: number; } export interface StreamOptions { // Opt out of this client's own retry. Exactly one caller wants that: the // download layer, which retries the request, the stream consumption and // the decryption as one unit. Leaving both layers enabled would multiply // the budgets — four attempts each becoming sixteen requests per file. retry?: boolean; } // An abort signal that fires once `ms` pass without a call to `restart`. It // aborts with a `TimeoutError`, the same reason `AbortSignal.timeout()` gives, // so the retry classifier treats an idle download exactly as it treats any // other deadline. `stop` must be called when the download ends. The timer is // unref'd, so even one left running never keeps the process alive. const idleDeadline = (ms: number) => { const controller = new AbortController(); let timer: ReturnType | undefined; const stop = (): void => clearTimeout(timer); const restart = (): void => { stop(); timer = setTimeout(() => { controller.abort( new DOMException( `download stalled: no bytes received for ${ms} ms`, "TimeoutError", ), ); }, ms); timer.unref(); }; restart(); return { signal: controller.signal, restart, stop }; }; // Enforce the idle deadline over a response body, not merely over its headers. // // `getFileStream` returns as soon as headers arrive; the bytes are pulled // later, in the download layer. Whether the signal passed to `fetch` also // tears down the body afterwards is up to the fetch implementation, so this // wrapper makes it a property of quak instead: every read races the signal, // each chunk that arrives restarts the deadline, and an abort errors the // stream with the abort reason — which the retry classifier recognises. const deadlineStream = ( body: ReadableStream, deadline: ReturnType, ): ReadableStream => { const { signal } = deadline; const reader = body.getReader(); let rejectOnAbort: (reason: unknown) => void = () => undefined; const aborted = new Promise((_resolve, reject) => { rejectOnAbort = reject; }); // The abort may fire when nothing is awaiting `aborted` — after the body // has been read in full, say. Without this, that rejection would surface // as an unhandled rejection and take the process down. void aborted.catch(() => undefined); const onAbort = (): void => rejectOnAbort(signal.reason); if (signal.aborted) onAbort(); else signal.addEventListener("abort", onAbort, { once: true }); const release = (): void => { deadline.stop(); signal.removeEventListener("abort", onAbort); }; return new ReadableStream({ async pull(controller) { try { const next = await Promise.race([reader.read(), aborted]); if (next.done) { release(); controller.close(); return; } deadline.restart(); controller.enqueue(next.value); } catch (err) { release(); await reader.cancel(err).catch(() => undefined); throw err; } }, async cancel(reason) { release(); await reader.cancel(reason); }, }); }; // The one place a request URL is built. `origin` may carry a base path (a // self-hosted server behind a prefix) and may end in a slash; `path` may or // may not start with one. Query parameters go only through `query`, which // percent-encodes them: a `?` or `#` in `path` is an error, because // `new URL` would otherwise quietly treat what follows as something else. const buildURL = ( origin: string, path: string, query?: Record, ): string => { if (path.includes("?") || path.includes("#")) { throw new Error( `request path must not contain "?" or "#"; pass query parameters separately: ${path}`, ); } const url = new URL( `${origin.replace(/\/+$/, "")}/${path.replace(/^\/+/, "")}`, ); for (const [k, v] of Object.entries(query ?? {})) { if (v !== undefined) url.searchParams.set(k, String(v)); } return url.href; }; export class ApiClient { private readonly apiOrigin: string; private readonly isCustomOrigin: boolean; private readonly filesOrigin: string; private readonly thumbsOrigin: string; private readonly _fetch: typeof globalThis.fetch; private readonly retry: ResolvedRetryOptions; private readonly requestTimeoutMs: number; private readonly downloadTimeoutMs: number; private token: string | undefined; constructor(opts?: ApiClientOptions) { this.apiOrigin = (opts?.apiOrigin ?? DEFAULT_API_ORIGIN).replace( /\/+$/, "", ); this.isCustomOrigin = this.apiOrigin !== DEFAULT_API_ORIGIN; this.filesOrigin = (opts?.filesOrigin ?? DEFAULT_FILES_ORIGIN).replace( /\/+$/, "", ); this.thumbsOrigin = ( opts?.thumbsOrigin ?? DEFAULT_THUMBS_ORIGIN ).replace(/\/+$/, ""); this._fetch = opts?.fetch ?? globalThis.fetch; this.retry = resolveRetryOptions(opts?.retry); this.requestTimeoutMs = opts?.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; this.downloadTimeoutMs = opts?.downloadTimeoutMs ?? DEFAULT_DOWNLOAD_TIMEOUT_MS; this.token = opts?.authToken; } setAuthToken(token: string): void { this.token = token; } clearAuthToken(): void { this.token = undefined; } getAuthToken(): string | undefined { return this.token; } // The policy this client was configured with, so that a caller wrapping a // whole operation in its own `withRetry` — the download layer — runs under // the same settings rather than under the library defaults. // A copy, so the caller cannot change this client's settings through it. getRetryOptions(): ResolvedRetryOptions { return { ...this.retry }; } private headers(extra?: Record): Record { const h: Record = { "X-Client-Package": CLIENT_PACKAGE, ...extra, }; if (this.token) { h["X-Auth-Token"] = this.token; } return h; } private async throwIfError(resp: Response): Promise { if (resp.ok) return; const requestID = resp.headers.get("x-request-id") ?? undefined; let body: unknown; let code: string | undefined; let message = `HTTP ${resp.status}`; try { const ct = resp.headers.get("content-type") ?? ""; if (ct.includes("application/json")) { body = await resp.json(); if ( body && typeof body === "object" && "code" in body && typeof (body as Record).code === "string" ) { code = (body as Record).code; } if ( body && typeof body === "object" && "message" in body && typeof (body as Record).message === "string" ) { message = (body as Record).message!; } } else { body = await resp.text(); } } catch { // body parsing failed; proceed with what we have } throw new ApiError(message, resp.status, { code, requestID, body }); } async getJSON( path: string, query?: Record, ): Promise { const url = buildURL(this.apiOrigin, path, query); // A GET changes nothing, so it is retried under the full policy. return withRetry(async () => { const resp = await this._fetch(url, { method: "GET", headers: this.headers(), signal: AbortSignal.timeout(this.requestTimeoutMs), }); await this.throwIfError(resp); return (await resp.json()) as T; }, this.retry); } async postJSON(path: string, body: unknown): Promise { const url = buildURL(this.apiOrigin, path); // Not idempotent: a POST is replayed only when `isSafeToReplay` // says no request byte can have reached the server. The endpoints // this covers are listed in the README under "Endpoints used". // // Redirects are not followed. The origin has already received the // request when it answers with one, so a connection refused by the // redirect target would look replay-safe when it is not. The API has // no legitimate redirect, so one surfaces as an `ApiError` with its // 3xx status, which is not retried. return withRetry( async () => { const resp = await this._fetch(url, { method: "POST", headers: this.headers({ "Content-Type": "application/json", }), body: JSON.stringify(body), redirect: "manual", signal: AbortSignal.timeout(this.requestTimeoutMs), }); await this.throwIfError(resp); return (await resp.json()) as T; }, { ...this.retry, isRetryable: isSafeToReplay }, ); } async getFileStream( fileID: number, opts?: StreamOptions, ): Promise> { const url = this.isCustomOrigin ? buildURL(this.apiOrigin, `/files/download/${fileID}`) : buildURL(this.filesOrigin, "/", { fileID }); return this.streamRequest(url, opts); } async getUploadURL( contentLength: number, contentMD5: string, ): Promise<{ objectKey: string; url: string }> { return this.postJSON("/files/upload-url", { contentLength, contentMD5, }); } async putFile(presignedURL: string, data: Uint8Array): Promise { // Idempotent despite being a write: a presigned PUT stores one whole // object at one key in one request, so replaying it either overwrites // the same bytes or lands them for the first time. There is no partial // state to protect, hence the full policy rather than the POST rule. await withRetry(async () => { const resp = await this._fetch(presignedURL, { method: "PUT", headers: { "Content-Type": "application/octet-stream", "Content-Length": String(data.length), }, body: data, signal: AbortSignal.timeout(this.requestTimeoutMs), }); if (!resp.ok) { // An ApiError, not a bare Error: without the status on the // error the upload path cannot be classified at all, and a // 503 from S3 would be indistinguishable from a bug. throw new ApiError( `PUT to presigned URL failed: HTTP ${resp.status}`, resp.status, ); } }, this.retry); } async putJSON(path: string, body: unknown): Promise { const url = buildURL(this.apiOrigin, path); // Same replay and redirect rules as `postJSON`, for the same reasons. return withRetry( async () => { const resp = await this._fetch(url, { method: "PUT", headers: this.headers({ "Content-Type": "application/json", }), body: JSON.stringify(body), redirect: "manual", signal: AbortSignal.timeout(this.requestTimeoutMs), }); await this.throwIfError(resp); return (await resp.json()) as T; }, { ...this.retry, isRetryable: isSafeToReplay }, ); } async updateThumbnail( fileID: number, objectKey: string, decryptionHeader: string, ): Promise { await this.putJSON("/files/thumbnail", { fileID, thumbnail: { objectKey, decryptionHeader }, }); } async getThumbnailStream( fileID: number, opts?: StreamOptions, ): Promise> { const url = this.isCustomOrigin ? buildURL(this.apiOrigin, `/files/preview/${fileID}`) : buildURL(this.thumbsOrigin, "/", { fileID }); return this.streamRequest(url, opts); } private async streamRequest( url: string, opts?: StreamOptions, ): Promise> { const once = async (): Promise> => { // A fresh deadline per attempt. It also covers the wait for the // headers, when no bytes have arrived either. const deadline = idleDeadline(this.downloadTimeoutMs); try { const resp = await this._fetch(url, { method: "GET", headers: this.headers(), signal: deadline.signal, }); await this.throwIfError(resp); if (!resp.body) { // Carries the status, and is not retryable: a response // that arrived without a body is malformed, and asking // again produces the same malformed response. throw new ApiError("response body is null", resp.status); } return deadlineStream(resp.body, deadline); } catch (err) { deadline.stop(); throw err; } }; return opts?.retry === false ? once() : withRetry(once, this.retry); } }