feat(sync): harden the Hiboutik sync, share rate limits across instances, update dependencies (#4)
This commit is contained in:
@@ -1,12 +1,49 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { HttpService } from '@nestjs/axios';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { isAxiosError } from 'axios';
|
||||
import { firstValueFrom } from 'rxjs';
|
||||
import { z } from 'zod';
|
||||
import { parseHiboutik } from '../models/hiboutikDTO/hiboutik.schema';
|
||||
|
||||
/** Results per page of every paginated Hiboutik endpoint (see "Results per page" in the API docs). */
|
||||
const HIBOUTIK_PAGE_SIZE = 250;
|
||||
|
||||
/** Low-level HTTP client for the Hiboutik REST API: base URL and Basic auth credentials come from the environment. */
|
||||
/** Retries of a request answered 429 (too many requests), with exponential backoff: 2 s, 4 s, 8 s. */
|
||||
const RATE_LIMIT_RETRIES = 3;
|
||||
const RATE_LIMIT_BASE_DELAY_MS = 2_000;
|
||||
/** Cap of the wait asked by a `Retry-After` header: beyond, waiting holds the sync for nothing (monthly quota). */
|
||||
const RATE_LIMIT_MAX_DELAY_MS = 60_000;
|
||||
|
||||
/**
|
||||
* Hiboutik still answers 429 after the retries: the rate limit (30 concurrent calls) or the monthly quota (10,000
|
||||
* calls per shop) is reached. The synchronization must stop: every next call would fail the same way.
|
||||
*/
|
||||
export class HiboutikRateLimitError extends Error {
|
||||
constructor(path: string) {
|
||||
super(`Hiboutik rate limit reached on ${path} (still 429 after ${RATE_LIMIT_RETRIES} retries): monthly quota exhausted?`);
|
||||
this.name = 'HiboutikRateLimitError';
|
||||
}
|
||||
}
|
||||
|
||||
/** Rethrows a {@link HiboutikRateLimitError}: for the per-item `catch` blocks of the sync, which must not swallow it. */
|
||||
export function rethrowIfRateLimited(error: unknown): void {
|
||||
if (error instanceof HiboutikRateLimitError) throw error;
|
||||
}
|
||||
|
||||
/** Wait asked by a `Retry-After` header (seconds or HTTP date), if any. */
|
||||
function retryAfterMs(header: unknown): number | undefined {
|
||||
if (typeof header !== 'string' || header.trim() === '') return undefined;
|
||||
const seconds = Number(header);
|
||||
if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);
|
||||
const date = Date.parse(header);
|
||||
return Number.isNaN(date) ? undefined : Math.max(0, date - Date.now());
|
||||
}
|
||||
|
||||
/**
|
||||
* Low-level HTTP client for the Hiboutik REST API: base URL and Basic auth credentials come from the environment.
|
||||
* Every answer is validated against a zod schema (see `hiboutik.schema.ts`): a malformed one throws, like an HTTP error.
|
||||
*/
|
||||
@Injectable()
|
||||
export class HiboutikClient {
|
||||
private readonly baseUrl: string;
|
||||
@@ -22,45 +59,109 @@ export class HiboutikClient {
|
||||
this.baseUrl = this.configService.get<string>('HIBOUTIK_BASE_URL') ?? '';
|
||||
}
|
||||
|
||||
/** GET on a Hiboutik path (relative to `HIBOUTIK_BASE_URL`), returns the response body. */
|
||||
async get<T>(path: string, params?: Record<string, unknown>): Promise<T> {
|
||||
const response = await firstValueFrom(
|
||||
this.httpService.get<T>(`${this.baseUrl}${path}`, {
|
||||
headers: { Authorization: this.authHeader },
|
||||
params,
|
||||
}),
|
||||
);
|
||||
return response.data;
|
||||
/**
|
||||
* GET on a Hiboutik path (relative to `HIBOUTIK_BASE_URL`), returns the raw response body. A 429 is retried with
|
||||
* backoff (or after the `Retry-After` delay), then turns into a {@link HiboutikRateLimitError}.
|
||||
*/
|
||||
private async get(path: string, params?: Record<string, unknown>): Promise<unknown> {
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
try {
|
||||
const response = await firstValueFrom(
|
||||
this.httpService.get<unknown>(`${this.baseUrl}${path}`, {
|
||||
headers: { Authorization: this.authHeader },
|
||||
params,
|
||||
}),
|
||||
);
|
||||
return response.data;
|
||||
} catch (error) {
|
||||
if (!isAxiosError(error) || error.response?.status !== 429) throw error;
|
||||
if (attempt >= RATE_LIMIT_RETRIES) throw new HiboutikRateLimitError(path);
|
||||
|
||||
const delay = retryAfterMs(error.response.headers['retry-after']) ?? RATE_LIMIT_BASE_DELAY_MS * 2 ** attempt;
|
||||
await this.sleep(Math.min(delay, RATE_LIMIT_MAX_DELAY_MS));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Separate method so that tests can skip the waits. */
|
||||
protected sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
/**
|
||||
* GET on a list endpoint. HTTP errors (4xx/5xx) throw, a 2xx answer without body is an empty list, and any other
|
||||
* non-list body throws: an empty list triggers deletions in the sync, so it must never be inferred from a malformed answer.
|
||||
* non-list body or invalid item throws: an empty list triggers deletions in the sync, so it must never be inferred
|
||||
* from a malformed answer.
|
||||
*/
|
||||
async getList<T>(path: string, params?: Record<string, unknown>): Promise<T[]> {
|
||||
const data = await this.get<T[] | '' | null | undefined>(path, params);
|
||||
async getList<T extends z.ZodType>(path: string, itemSchema: T, params?: Record<string, unknown>): Promise<z.output<T>[]> {
|
||||
const data = await this.get(path, params);
|
||||
if (data === '' || data == null) return [];
|
||||
if (!Array.isArray(data)) throw new Error(`Unexpected answer from Hiboutik ${path}: a list was expected`);
|
||||
return data;
|
||||
return parseHiboutik(z.array(itemSchema), data, path);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches every page of a paginated endpoint (`p` param, starting at 1) and concatenates the results.
|
||||
* Only for paginated endpoints: on a non-paginated one, a full first page would loop forever on the same results.
|
||||
*/
|
||||
async getAllPages<T>(path: string, baseParams: Record<string, unknown> = {}): Promise<T[]> {
|
||||
const results: T[] = [];
|
||||
async getAllPages<T extends z.ZodType>(path: string, itemSchema: T, baseParams: Record<string, unknown> = {}): Promise<z.output<T>[]> {
|
||||
return (await this.walkPages(path, itemSchema, baseParams)).items;
|
||||
}
|
||||
|
||||
/**
|
||||
* Same as {@link getAllPages}, for a list whose content drives deletions. An item created or deleted in Hiboutik while
|
||||
* the pages are walked shifts the following pages, so an item can be skipped. When the list spans several pages, it
|
||||
* is walked a second time and both walks are merged by id (the second one wins): an item is missing from the result
|
||||
* only if both walks missed it. A list that fits in one page costs no extra call.
|
||||
*/
|
||||
async getAllPagesForSync<T extends z.ZodType>(
|
||||
path: string,
|
||||
itemSchema: T,
|
||||
idOf: (item: z.output<T>) => number,
|
||||
baseParams: Record<string, unknown> = {},
|
||||
): Promise<z.output<T>[]> {
|
||||
const firstWalk = await this.walkPages(path, itemSchema, baseParams);
|
||||
if (firstWalk.nbPages <= 1) return firstWalk.items;
|
||||
|
||||
const secondWalk = await this.walkPages(path, itemSchema, baseParams);
|
||||
const byId = new Map<number, z.output<T>>();
|
||||
for (const item of [...firstWalk.items, ...secondWalk.items]) byId.set(idOf(item), item);
|
||||
return [...byId.values()];
|
||||
}
|
||||
|
||||
/** Walks the pages of a paginated endpoint. `nbPages` counts the pages holding results. */
|
||||
private async walkPages<T extends z.ZodType>(
|
||||
path: string,
|
||||
itemSchema: T,
|
||||
baseParams: Record<string, unknown>,
|
||||
): Promise<{ items: z.output<T>[]; nbPages: number }> {
|
||||
const items: z.output<T>[] = [];
|
||||
let page = 1;
|
||||
|
||||
while (true) {
|
||||
const pageData = await this.getList<T>(path, { ...baseParams, p: page });
|
||||
const pageData = await this.getList(path, itemSchema, { ...baseParams, p: page });
|
||||
if (pageData.length === 0) break;
|
||||
results.push(...pageData);
|
||||
items.push(...pageData);
|
||||
// A page smaller than the page size is the last one: no need to request an empty page
|
||||
if (pageData.length < HIBOUTIK_PAGE_SIZE) break;
|
||||
page++;
|
||||
}
|
||||
|
||||
return results;
|
||||
return { items, nbPages: Math.ceil(items.length / HIBOUTIK_PAGE_SIZE) };
|
||||
}
|
||||
|
||||
/**
|
||||
* GET on a single-resource endpoint (Hiboutik answers with a one-element array).
|
||||
* @returns `null` when Hiboutik says the resource does not exist: a 404, or a 2xx answer that is empty.
|
||||
* @throws on any other HTTP error or malformed answer: an unknown state is never taken for a missing resource.
|
||||
*/
|
||||
async getOne<T extends z.ZodType>(path: string, schema: T): Promise<z.output<T> | null> {
|
||||
try {
|
||||
const [item] = await this.getList(path, schema);
|
||||
return item ?? null;
|
||||
} catch (error) {
|
||||
if (isAxiosError(error) && error.response?.status === 404) return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user