quak/src/api/client.ts

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<typeof setTimeout> | 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<Uint8Array>,
    deadline: ReturnType<typeof idleDeadline>,
): ReadableStream<Uint8Array> => {
    const { signal } = deadline;
    const reader = body.getReader();
    let rejectOnAbort: (reason: unknown) => void = () => undefined;
    const aborted = new Promise<never>((_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<Uint8Array>({
        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, string | number | undefined>,
): 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<string, string>): Record<string, string> {
        const h: Record<string, string> = {
            "X-Client-Package": CLIENT_PACKAGE,
            ...extra,
        };
        if (this.token) {
            h["X-Auth-Token"] = this.token;
        }
        return h;
    }

    private async throwIfError(resp: Response): Promise<void> {
        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<string, unknown>).code === "string"
                ) {
                    code = (body as Record<string, string>).code;
                }
                if (
                    body &&
                    typeof body === "object" &&
                    "message" in body &&
                    typeof (body as Record<string, unknown>).message ===
                        "string"
                ) {
                    message = (body as Record<string, string>).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<T>(
        path: string,
        query?: Record<string, string | number | undefined>,
    ): Promise<T> {
        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<T>(path: string, body: unknown): Promise<T> {
        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<ReadableStream<Uint8Array>> {
        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<void> {
        // 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<T>(path: string, body: unknown): Promise<T> {
        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<void> {
        await this.putJSON("/files/thumbnail", {
            fileID,
            thumbnail: { objectKey, decryptionHeader },
        });
    }

    async getThumbnailStream(
        fileID: number,
        opts?: StreamOptions,
    ): Promise<ReadableStream<Uint8Array>> {
        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<ReadableStream<Uint8Array>> {
        const once = async (): Promise<ReadableStream<Uint8Array>> => {
            // 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);
    }
}